PySpark：将含字典列表的DataFrame列解析为指定表格结构

嘿，我来帮你搞定这个PySpark的解析问题！你需要把每行payload里的字符串格式字典列表，转成以各个key为列、对应value填充的结构化DataFrame对吧？下面是一步步的实现方案，附带代码示例：

1. 先准备测试数据 #

首先模拟你的输入DataFrame，方便后续演示效果：

from pyspark.sql import Row
from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, from_json, map_keys, map_values, col, first, regexp_replace, monotonically_increasing_id

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

# 模拟你的原始数据
data = [
    Row(payload=u"[{'key1':'value1'},{'key2':'value2'},{'key3':'value3'}]"),
    Row(payload=u"[{'key1':'valueA'},{'key3':'valueC'},{'key4':'valueD'}]")
]
df = spark.createDataFrame(data)
df.show(truncate=False)

2. 把字符串转成PySpark能识别的数组类型 #

你的payload是单引号格式的字符串字典列表，但PySpark的from_json默认只认双引号的标准JSON，所以第一步得把单引号替换成双引号，再解析成数组类型：

from pyspark.sql.types import ArrayType, MapType, StringType

# 替换单引号为双引号，修复JSON格式
df_clean = df.withColumn("payload_clean", regexp_replace(col("payload"), "'", "\""))
# 解析为数组<map<string, string>>类型（数组里每个元素是键值对字典）
df_parsed = df_clean.withColumn("payload_array", from_json(col("payload_clean"), ArrayType(MapType(StringType(), StringType()))))

df_parsed.select("payload_array").show(truncate=False)

3. 炸开数组里的每个字典 #

用explode函数把数组中的每个字典拆成单独的行，这样每行就对应一个键值对了：

df_exploded = df_parsed.withColumn("single_map", explode(col("payload_array")))
df_exploded.select("single_map").show(truncate=False)

4. 提取每个字典的key和value #

因为每个字典只有一个键值对，我们可以用map_keys和map_values分别取出key和value（取索引0是因为每个map只有一对）：

df_key_value = df_exploded.withColumn("key", map_keys(col("single_map"))[0]) \
                          .withColumn("value", map_values(col("single_map"))[0])

df_key_value.select("key", "value").show()

5. Pivot转成结构化列 #

最后用pivot把key转成列，并用first聚合对应的值——这里要先给每行加个唯一ID，避免pivot时把不同行的数据合并：

# 给每行加唯一标识，防止pivot时混淆不同行的数据
df_with_id = df_key_value.withColumn("row_id", monotonically_increasing_id())

# 按row_id分组，pivot key列，用first取对应的value
result_df = df_with_id.groupBy("row_id").pivot("key").agg(first("value"))

# 不需要row_id的话可以删掉
result_df = result_df.drop("row_id")

result_df.show()

最终输出效果 #

运行完上面的代码后，你会得到这样的结构化DataFrame：

+-------+-------+-------+-------+
|  key1 |  key2 |  key3 |  key4 |
+-------+-------+-------+-------+
|value1 |value2 |value3 |   null|
|valueA |   null|valueC |valueD |
+-------+-------+-------+-------+

一些额外提醒 #

• 如果你的字典里有多个键值对（不是单个），那得调整提取逻辑：比如用explode(map_entries(col("single_map")))把每个map的键值对也炸开，再处理。
• 如果payload字符串里有特殊转义字符，可能需要调整regexp_replace的规则，保证替换后是合法的JSON。
• 聚合函数first可以根据需求替换：比如同一行同一key有多个值，用collect_list保留所有值，或者用max/min取极值。

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