PySpark字典转DataFrame：求优雅高效实现方案（指定输入输出）

1. 初始化SparkSession #

先完成Spark环境初始化：

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, collect_list

spark = SparkSession.builder.appName("DictToSparkDF").getOrCreate()

2. 生成中间形式DataFrame #

先将输入字典转为包含keys和对应值数组的DataFrame，再通过explode展开数组得到中间结果：

data = {"key1": ["val1", "val2", "val3"], "key2": ["val3", "val4", "val5"]}

# 创建存储key与值数组的初始DF
initial_df = spark.createDataFrame(data.items(), schema=["keys", "vals_array"])
# 展开值数组，得到中间形式DF
intermediate_df = initial_df.select("keys", explode("vals_array").alias("vals"))

# 查看中间结果
intermediate_df.show()

输出结果：

+----+----+
|keys|vals|
+----+----+
|key1|val1|
|key1|val2|
|key1|val3|
|key2|val3|
|key2|val4|
|key2|val5|
+----+----+

3. 生成最终目标DataFrame #

基于中间DF按vals分组，用collect_list聚合对应keys：

final_df = intermediate_df.groupBy("vals").agg(collect_list("keys").alias("keys"))

# 查看最终结果
final_df.show()

输出结果：

+----+------------+
|vals|        keys|
+----+------------+
|val1|      [key1]|
|val2|      [key1]|
|val3|[key1, key2]|
|val4|      [key2]|
|val5|      [key2]|
+----+------------+

方案优势 #

• 全程依赖Spark原生API处理，避免Python端循环操作，适配大数据场景下的高效计算
• explode、collect_list均为Spark内置高阶函数，性能优于自定义逻辑

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