{"id":88180,"date":"2024-12-14T06:54:42","date_gmt":"2024-12-14T03:24:42","guid":{"rendered":"https:\/\/nabfollower.com\/blog\/azure-synapse-pyspark-toolbox-002-dataframe-transformation-500f\/"},"modified":"2024-12-14T06:54:42","modified_gmt":"2024-12-14T03:24:42","slug":"azure-synapse-pyspark-toolbox-002-dataframe-transformation-500f","status":"publish","type":"post","link":"https:\/\/nabfollower.com\/blog\/azure-synapse-pyspark-toolbox-002-dataframe-transformation-500f\/","title":{"rendered":"Azure Synapse PySpark Toolbox 002: DataFrame Transformation"},"content":{"rendered":"

Summarize this content to 400 words in Persian Lang
\n \u0645\u062d\u062a\u0648\u06cc\u0627\u062a \u062c\u0639\u0628\u0647 \u0627\u0628\u0632\u0627\u0631<\/p>\n

\u062a\u0628\u062f\u06cc\u0644 \u067e\u0627\u06cc\u0647 DataFrame \u0633\u062a\u0648\u0646 \u0647\u0627\u06cc \u062a\u062c\u0645\u0639\u06cc \u0631\u0627 \u0627\u0632 DataFrame \u0641\u0631\u0632\u0646\u062f \u0627\u06cc\u062c\u0627\u062f \u06a9\u0646\u06cc\u062f<\/p>\n

\u0645\u0634\u062e\u0635\u0627\u062a \u0645\u062d\u06cc\u0637\u06cc<\/p>\n

Azure Synapse Runtime \u0628\u0631\u0627\u06cc Apache Spark 3.4
\nAzure Data Lake Storage
\n\u062e\u0631\u06a9 \u06a9\u0644\u06cc\u062f \u0644\u0627\u062c\u0648\u0631\u062f\u06cc<\/p>\n

\u0648\u0627\u0628\u0633\u062a\u06af\u06cc \u0647\u0627\u06cc \u0648\u0627\u0631\u062f\u0627\u062a\u06cc<\/p>\n

import requests
\nfrom pyspark.sql import types as T, functions as F
\nimport json
\nimport datetime
\nimport logging<\/p>\n

\u0648\u0627\u0631\u062f \u062d\u0627\u0644\u062a \u062a\u0645\u0627\u0645 \u0635\u0641\u062d\u0647 \u0634\u0648\u06cc\u062f<\/p>\n

\u0627\u0632 \u062d\u0627\u0644\u062a \u062a\u0645\u0627\u0645 \u0635\u0641\u062d\u0647 \u062e\u0627\u0631\u062c \u0634\u0648\u06cc\u062f<\/p>\n

\u062a\u0628\u062f\u06cc\u0644 \u067e\u0627\u06cc\u0647 DataFrame <\/p>\n

\u062a\u0628\u062f\u06cc\u0644 \u067e\u0627\u06cc\u0647 DataFrame (\u0627\u0641\u0632\u0648\u062f\u0646 \u0633\u062a\u0648\u0646\u200c\u0647\u0627\u06cc \u062c\u062f\u06cc\u062f\u060c \u062a\u063a\u06cc\u06cc\u0631 \u0646\u0627\u0645\u060c \u067e\u0631 \u06a9\u0631\u062f\u0646 \u0645\u0642\u0627\u062f\u06cc\u0631 na\u060c \u0641\u06cc\u0644\u062a\u0631 \u0631\u062f\u06cc\u0641\u200c\u0647\u0627\u060c \u0627\u0646\u062a\u062e\u0627\u0628 \u0633\u062a\u0648\u0646\u200c\u0647\u0627\u060c \u0631\u0647\u0627 \u06a9\u0631\u062f\u0646 \u0645\u0648\u0627\u0631\u062f \u062a\u06a9\u0631\u0627\u0631\u06cc\u060c \u0628\u0631\u06af\u0631\u062f\u0627\u0646\u062f\u0646 \u06cc\u06a9 \u0634\u06cc PySpark Dataframe)<\/p>\n

def df_transform_basic(
\n df_input, #DataFrame
\n step1_new_cols, #List of Dictionaries [{‘new_col_name’: ‘xxx’, ‘new_col_expr’: ‘expression string for F.expr() function’}]\n step2_col_name_mapping, #Dictionary of old-new names mapping
\n step3_na_values,
\n step4_filter_expr, #String parameter for df.filter() function
\n step5_result_col_list, #List of column names
\n step6_drop_dup, #True\/False
\n timestamp_name
\n):
\n df_output = df_input<\/p>\n

if step1_new_cols:
\n for val in step1_new_cols:
\n df_output = df_output.withColumn(val[‘new_col_name’], F.expr(val[‘new_col_expr’]))<\/p>\n

if step2_col_name_mapping:
\n df_output = df_output.withColumnsRenamed(step2_col_name_mapping)<\/p>\n

if step3_na_values:
\n df_output = df_output.na.fill(step3_na_values)<\/p>\n

if step4_filter_expr:
\n df_output = df_output.filter(step4_filter_expr)<\/p>\n

if step5_result_col_list:
\n df_output = df_output.select(*step5_result_col_list)<\/p>\n

if step6_drop_dup:
\n df_output = df_output.dropDuplicates()<\/p>\n

if timestamp_name:
\n df_output = df_output.withColumn(timestamp_name, F.lit(F.current_timestamp()))<\/p>\n

return df_output<\/p>\n

\u0648\u0627\u0631\u062f \u062d\u0627\u0644\u062a \u062a\u0645\u0627\u0645 \u0635\u0641\u062d\u0647 \u0634\u0648\u06cc\u062f<\/p>\n

\u0627\u0632 \u062d\u0627\u0644\u062a \u062a\u0645\u0627\u0645 \u0635\u0641\u062d\u0647 \u062e\u0627\u0631\u062c \u0634\u0648\u06cc\u062f<\/p>\n

\u0628\u0627\u0632\u06af\u0634\u062a \u0628\u0647 \u0628\u0627\u0644\u0627<\/p>\n

\u0633\u062a\u0648\u0646 \u0647\u0627\u06cc \u062a\u062c\u0645\u0639\u06cc \u0631\u0627 \u0627\u0632 DataFrame \u0641\u0631\u0632\u0646\u062f \u0627\u06cc\u062c\u0627\u062f \u06a9\u0646\u06cc\u062f <\/p>\n

\u0633\u062a\u0648\u0646 \u0647\u0627\u06cc \u062a\u062c\u0645\u0639\u06cc \u0631\u0627 \u0628\u0631 \u0627\u0633\u0627\u0633 \u062f\u0627\u062f\u0647 \u0647\u0627\u06cc DataFrame \u0641\u0631\u0632\u0646\u062f \u0627\u06cc\u062c\u0627\u062f \u06a9\u0646\u06cc\u062f \u0648 DataFrame \u0628\u0647 \u0631\u0648\u0632 \u0634\u062f\u0647 \u0648\u0627\u0644\u062f\u06cc\u0646 \u0631\u0627 \u0628\u0631\u0645\u06cc \u06af\u0631\u062f\u0627\u0646\u062f. \u067e\u06cc\u06a9\u0631\u0628\u0646\u062f\u06cc \u0647\u0631 \u062a\u062c\u0645\u0639 \u062f\u0631 \u0634\u06cc\u0621 \u062f\u06cc\u06a9\u0634\u0646\u0631\u06cc summary_config \u062a\u0639\u0631\u06cc\u0641 \u0634\u062f\u0647 \u0627\u0633\u062a.<\/p>\n

def df_summary(
\n parent_df, #DataFrame
\n child_df, #DataFrame
\n summary_config
\n):<\/p>\n

# Example of summary_config
\n# summary_config=[{‘agg’ : {‘RevisionNum’: ‘count’}, ‘agg_result_col_name’ : ‘CountOfRevisions’, ‘na_fill_value’ : 0, ‘join_on’ = [‘Contract’, ‘RevisionNum’]},
\n# {‘agg’ : {‘RevisionNum’: ‘max’}, ‘agg_result_col_name’ : ‘RecentRevNum’, ‘na_fill_value’ : 0, ‘join_on’ = [‘Contract’, ‘RevisionNum’]}]\n

if summary_config:
\n for summary_item in summary_config:
\n agg_col = list(summary_item[‘agg’])[0]\n agg_fun = summary_item[‘agg’][agg_col]\n child_df_grouped = child_df.groupBy(*join_on).agg(summary_item[‘agg’]).withColumnRenamed(f”{agg_fun}({agg_col})”, summary_item[‘agg_result_col_name’])
\n parent_df = parent_df.join(child_df_grouped, summary_item[‘agg’][‘join_on’], ‘left’)
\n na_value = {}
\n na_value[summary_item[‘agg_result_col_name’]] = summary_item[‘na_fill_value’]\n parent_df = parent_df.na.fill(na_value)<\/p>\n

return parent_df<\/p>\n

\u0648\u0627\u0631\u062f \u062d\u0627\u0644\u062a \u062a\u0645\u0627\u0645 \u0635\u0641\u062d\u0647 \u0634\u0648\u06cc\u062f<\/p>\n

\u0627\u0632 \u062d\u0627\u0644\u062a \u062a\u0645\u0627\u0645 \u0635\u0641\u062d\u0647 \u062e\u0627\u0631\u062c \u0634\u0648\u06cc\u062f<\/p>\n

\u0628\u0627\u0632\u06af\u0634\u062a \u0628\u0647 \u0628\u0627\u0644\u0627<\/p>\n

\n

\u0645\u062d\u062a\u0648\u06cc\u0627\u062a \u062c\u0639\u0628\u0647 \u0627\u0628\u0632\u0627\u0631<\/p>\n\n

\u062a\u0628\u062f\u06cc\u0644 \u067e\u0627\u06cc\u0647 DataFrame
\u0633\u062a\u0648\u0646 \u0647\u0627\u06cc \u062a\u062c\u0645\u0639\u06cc \u0631\u0627 \u0627\u0632 DataFrame \u0641\u0631\u0632\u0646\u062f \u0627\u06cc\u062c\u0627\u062f \u06a9\u0646\u06cc\u062f<\/p>\n\n

\u0645\u0634\u062e\u0635\u0627\u062a \u0645\u062d\u06cc\u0637\u06cc<\/p>\n