基于PySpark DataFrame生成全组合JSON结构对接ADF

实现步骤与代码示例 #

假设你的原始PySpark DataFrame名为input_df，直接基于以下代码完成需求：

from pyspark.sql.functions import col
import json

# 1. 提取三列的所有唯一取值组合
unique_combinations_df = input_df.select("databricksPath", "countryPartition", "yearPartition").dropDuplicates()

# 2. 将Spark行对象转为Python字典列表
combinations_list = [row.asDict() for row in unique_combinations_df.collect()]

# 3. 序列化为标准JSON数组并返回给ADF
json_output = json.dumps(combinations_list)
dbutils.notebook.exit(json_output)

关键说明 #

• dropDuplicates()：确保得到无重复的列值组合，避免Foreach活动处理冗余项。
• row.asDict()：将Spark的Row结构转为Python原生字典，为JSON序列化做准备。
• json.dumps()：生成ADF可直接解析的JSON数组格式。
• dbutils.notebook.exit()：将JSON结果返回给调用该Notebook的ADF活动，后续在Foreach活动的Items字段中用@activity('你的Notebook活动名称').output引用即可。

注意事项 #

• 如果原始DataFrame数据量过大，collect()可能引发Driver端内存溢出，需提前确认数据规模在合理范围（Foreach活动本身也不适合处理超大量迭代项）。
• 确保三列数据类型为字符串、数值等ADF可识别类型，避免JSON序列化异常。

内容的提问来源于stack exchange，提问作者coding