如何用PySpark将多列转换为单列复杂JSON并落地？

核心思路 #

完全利用PySpark内置的struct、array等结构化函数构建目标嵌套结构，避免使用Python UDF（UDF会引入JVM与Python进程间的序列化开销，性能远低于内置函数）；最后通过Spark原生JSON写入器输出单行无缩进的JSON Lines格式文件。

具体实现代码 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import struct, array

# 初始化SparkSession（若未初始化）
spark = SparkSession.builder.appName("FlatToNestedJson").getOrCreate()

# 假设输入DataFrame为df，已完成加载
# 构建目标嵌套结构
transformed_df = df.withColumn(
    "result",
    # 外层result数组，每行对应数组内一个对象
    array(
        struct(
            # pair数组包含两个子对象，映射对应列
            array(
                struct(df.col_a.alias("a"), df.col_b.alias("b")),
                struct(df.col_c.alias("c"), df.col_d.alias("d"))
            ).alias("pair")
        )
    )
).select("result")  # 仅保留最终需要的result列

# 写入输出文件：multiLine=false实现每行一个JSON，默认无缩进
transformed_df.write \
    .mode("overwrite")  # 可根据需求替换为append/ignore等模式
    .option("multiLine", "false") \
    .json("/your/output/path")

关键细节说明 #

1. 性能优势：struct、array是Spark底层优化的JVM级函数，处理大规模数据时性能比Python UDF高数倍，无需额外处理序列化逻辑。
2. 格式匹配：设置multiLine=false后，Spark会将每行数据输出为独立的JSON字符串（JSON Lines格式），且默认不添加缩进，完全符合“单行无缩进”要求。
3. 结构对应：通过嵌套的array和struct严格匹配目标JSON结构：
   • 最外层是result数组，包含一个对象
   • 该对象内的pair数组包含两个子对象，分别将col_a/col_b映射为a/b、col_c/col_d映射为c/d

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