PySpark如何将表字段按自定义键拼接为JSON新增列并导出

核心实现逻辑是先将原始字段按映射规则重命名后封装为结构化类型，再通过内置to_json函数转换为标准JSON字符串，可自动适配字段类型、处理转义规则，避免手动拼接的格式错误。

完整实现代码 #

from pyspark.sql import functions as F

# 定义原始字段到JSON键的映射关系
col_mapping = {
    "id": "internal_id",
    "product_id": "item_id",
    "agent": "associate_id",
    "order_date": "transaction_date"
}

# 生成重命名后的结构体字段列表
struct_fields = [F.col(old_col).alias(new_col) for old_col, new_col in col_mapping.items()]

# 新增目标JSON列
df_result = df.withColumn(
    "wanted_column",
    F.to_json(F.struct(*struct_fields))
)

导出JSON文件 #

仅保留目标列导出，每行对应一个业务需要的JSON对象：

df_result.select("wanted_column").write.mode("overwrite").json("你的输出目录路径")

常见问题说明 #

你之前尝试的方法效果不符合预期的原因：

• 手动concat拼接字符串：无法自动处理引号转义、字段类型适配，容易出现JSON格式错误
• map_from_entries方法：会统一将所有字段值转为字符串类型，无法保留数值型字段不带引号的格式，和预期输出不符

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