PySpark：将含Map列表的DataFrame转为可存CSV的结构化DataFrame

这里我们可以通过添加连续规则编号、展开数组、解析结构体三个步骤来实现需求，直接上可运行的代码：

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# 你的原始DataFrame
new_df = spark.createDataFrame([
    ([{'product_code': '12', 'color': 'red'}, {'product_code': '212', 'color': 'white'}], 7),
    ([{'product_code': '1112', 'color': 'black'}], 8),
    ([{'product_code': '212', 'color': 'blue'}], 3)
], ["items", "frequency"])

# 步骤1：给每一行原始数据生成连续的rule编号
# 用monotonically_increasing_id()保证行顺序，避免Spark默认分区打乱原始顺序
window_spec = Window.orderBy(F.monotonically_increasing_id())
df_with_rule = new_df.withColumn("rule", F.row_number().over(window_spec))

# 步骤2：展开items数组，把每个商品结构体拆成单独行
df_exploded = df_with_rule.withColumn("item_struct", F.explode("items"))

# 步骤3：从结构体中提取字段，得到最终结构化结果
final_df = df_exploded.select(
    "rule",
    F.col("item_struct.product_code").alias("product_code"),
    F.col("item_struct.color").alias("color")
)

# 查看输出结果
final_df.show()

运行后会输出你需要的结构化格式：

+----+------------+-----+
|rule|product_code|color|
+----+------------+-----+
|   1|          12|  red|
|   1|         212|white|
|   2|        1112|black|
|   3|         212| blue|
+----+------------+-----+

关键步骤说明 #

• 生成rule编号：用row_number()窗口函数配合monotonically_increasing_id()，既能保证编号连续，又能稳定维持你的原始数据行顺序（Spark DataFrame本身是无序的，直接排序可能依赖分区，这个方法更稳妥）。
• 展开数组：F.explode("items")会把数组里的每个商品结构体拆成独立行，同时保留对应的rule编号，实现"同一列表对应相同规则号"的要求。
• 解析结构体：通过.操作符直接从结构体字段中提取product_code和color，并重命名成你需要的字段名。

最后你可以用final_df.write.csv("你的输出路径", header=True)将结果保存为带表头的CSV文件。

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