如何用PySpark拆分含对象列表的DataFrame并生成指定列？

需求回顾 #

现有如下结构的PySpark DataFrame：

f1     |f2
=========
test   | [{"f3": 1, "f4": "f4_1" }, {"f3": 2, "f4": "f4_2" }] 

其中f2列是JSON对象组成的数组，需要转换为以下结构：

f3|f4    | temp_col
=========================
1 |"f4_1"| {"f1": "test"}
2 |"f4_2"| {"f1": "test"}

核心需求是：展开数组列、提取嵌套字段、生成包含原f1值的自定义列temp_col。

实现步骤 #

不需要转成Pandas处理，直接用PySpark原生API就能完成，步骤如下：

1. 导入PySpark函数库

   from pyspark.sql import functions as F
   

2. 展开数组列
   使用explode函数将f2数组中的每个元素拆分为单独的行，保留原f1列的值。

3. 提取嵌套字段
   从展开后的f2元素中提取f3和f4字段。

4. 生成自定义列temp_col
   使用create_map函数构造包含f1键值对的结构列。

完整代码示例 #

# 1. 创建测试DataFrame
data = [("test", [{"f3": 1, "f4": "f4_1"}, {"f3": 2, "f4": "f4_2"}])]
df = spark.createDataFrame(data, schema=["f1", "f2"])

# 2. 执行转换
result_df = df.select(
    F.explode("f2").alias("f2_exploded"),
    "f1"
).select(
    F.col("f2_exploded.f3").alias("f3"),
    F.col("f2_exploded.f4").alias("f4"),
    F.create_map(F.lit("f1"), F.col("f1")).alias("temp_col")
)

# 3. 查看结果
result_df.show(truncate=False)

结果说明 #

执行上述代码后，输出的DataFrame结构与需求完全一致：

+---+-----+-------------+
|f3 |f4   |temp_col     |
+---+-----+-------------+
|1  |f4_1 |{f1 -> test} |
|2  |f4_2 |{f1 -> test} |
+---+-----+-------------+

补充说明 #

如果f2列的类型不是ArrayType(StructType)，而是字符串格式的JSON数组，需要先使用from_json函数解析为数组结构，示例代码如下：

from pyspark.sql.types import ArrayType, StructType, IntegerType, StringType

# 定义f2列的Schema
f2_schema = ArrayType(StructType([
    StructField("f3", IntegerType()),
    StructField("f4", StringType())
]))

# 解析字符串为数组结构
df = df.withColumn("f2", F.from_json("f2", f2_schema))

解析完成后再执行前面的展开和提取步骤即可。

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