PySpark如何读取单文件多schema JSON并按id合并生成目标表

1. 读取JSON文件 #

PySpark读取多行格式的JSON数组时，需要开启multiLine参数，读取过程会自动合并所有出现过的字段，缺失字段会自动填充null值：

from pyspark.sql import SparkSession
from pyspark.sql.functions import first, element_at

# 初始化SparkSession
spark = SparkSession.builder.appName("MergeJsonById").getOrCreate()

# 替换为你的实际JSON文件路径
df = spark.read.json("test_data.json", multiLine=True)

2. 按id分组聚合合并字段 #

同一个id对应的多条记录里，每个字段最多只有一个非空值，用first函数忽略空值即可取出对应字段的值，同时提取platforms数组的第一个元素匹配预期输出格式：

result_df = df.groupBy("id").agg(
    first("price1", ignorenulls=True).alias("price1"),
    # 如果需要保留platforms的数组格式，直接写first("platforms", ignorenulls=True).alias("platforms")即可
    element_at(first("platforms", ignorenulls=True), 1).alias("platforms"),
    first("title", ignorenulls=True).alias("title")
)

3. 结果验证 #

# 打印表结构
result_df.printSchema()
# 展示输出结果
result_df.show(truncate=False)

兼容说明 #

如果使用的PySpark版本低于3.0，不支持first的ignorenulls参数，可以替换为max函数实现相同效果：

from pyspark.sql.functions import max

result_df = df.groupBy("id").agg(
    max("price1").alias("price1"),
    element_at(max("platforms"), 1).alias("platforms"),
    max("title").alias("title")
)

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