Spark DataFrame中含多逗号分隔JSON的列扁平化处理

我的Spark DataFrame包含id（整数）、name（字符串）和student_data（字符串）三列，其中student_data列的内容是多个以逗号分隔的键值对JSON对象组成的字符串，需要将该列的JSON数据扁平化为单独的列，最终得到以科目名为列、对应分数为值的结果。

数据示例 #

+--+------+---------------------------------------------------------------------------------------------------------------------------------------+
|id|name  |student_data                                                                                                                           |     
+--+------+---------------------------------------------------------------------------------------------------------------------------------------+
|11|stephy|{{"key":"hindi","value":{"hindi_mythology":80}},{"key":"social_science","value":{"civics":65}},{"key":"maths","value":{"geometry":70}}}
+--+------+---------------------------------------------------------------------------------------------------------------------------------------+

数据Schema #

root
|-- id : int
|-- name : string
|-- student_data : string

期望输出 #

+---+------+-----+--------------+-----+
| id|  name|hindi|social_science|maths|
+---+------+-----+--------------+-----+
| 11|stephy|   80|            65|   70|
+---+------+-----+--------------+-----+

解决方案（PySpark） #

步骤说明

1. 修正JSON格式：原student_data列的内容外层多了一对大括号，需要先去除首尾的大括号，将内容转换为合法的JSON数组格式。
2. 解析JSON数组：用from_json函数将修正后的字符串解析为Spark数组类型，数组元素包含key（科目名）和value（包含分数的嵌套JSON）。
3. 展开数组元素：使用explode函数将数组拆分为多行，每行对应一个科目数据。
4. 提取分数值：从value中提取分数（每个value仅包含一个键值对，直接取对应值即可）。
5. 透视表转换列：通过pivot将科目名转换为列，聚合得到每个id和name对应的各科目分数。

代码实现

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, explode, col, map_values, first

# 初始化Spark会话
spark = SparkSession.builder.appName("StudentDataFlatten").getOrCreate()

# 构造示例数据
data = [(11, "stephy", '{{"key":"hindi","value":{"hindi_mythology":80}},{"key":"social_science","value":{"civics":65}},{"key":"maths","value":{"geometry":70}}}')]
df = spark.createDataFrame(data, ["id", "name", "student_data"])

# 1. 修正JSON格式，转为合法数组
df_clean = df.withColumn(
    "student_data_array",
    from_json(
        # 去除首尾的大括号，包裹成数组
        col("student_data").substr(2, col("student_data").length() - 2),
        "array<struct<key:string,value:map<string,int>>>"
    )
)

# 2. 展开数组，拆分每个科目数据
df_exploded = df_clean.select("id", "name", explode(col("student_data_array")).alias("subject_info"))

# 3. 提取科目名和分数
df_extracted = df_exploded.select(
    "id",
    "name",
    col("subject_info.key").alias("subject"),
    # 从value的map中提取分数（取第一个值）
    map_values(col("subject_info.value"))[0].alias("score")
)

# 4. 透视表将科目转为列，聚合分数
df_result = df_extracted.groupBy("id", "name").pivot("subject").agg(first("score"))

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

关键说明

• 如果你的Spark版本低于3.0，没有map_values函数，可以改用get_json_object提取分数：

  get_json_object(col("subject_info.value").cast("string"), "$.*").cast("int").alias("score")
  

• 若student_data的原始格式是合法的JSON数组（即外层是[{}]而非{{}}），可以直接跳过格式修正步骤，直接解析即可。

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