如何在Spark中基于行数据动态添加新列？

针对你遇到的动态JSON列扩展问题，除了加唯一键关联的方式，还有这几种更直接的实现思路：

方法1：推断JSON统一Schema后用from_json展开 #

如果你的JSON数据虽然字段不固定，但可以先收集所有可能的字段生成统一Schema，再用from_json解析后直接展开：

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, MapType

# 1. 先收集所有JSON的字段，生成统一Schema
json_dicts = df.select(F.from_json(F.col("Col_C"), MapType(StringType(), StringType())).alias("json_data")) \
               .select("json_data.*").columns
# 构造StructType
json_schema = StructType([StructField(col, StringType(), nullable=True) for col in json_dicts])

# 2. 解析JSON并展开字段
parsed_df = df.withColumn("json_data", F.from_json(F.col("Col_C"), json_schema))
# 选择原列+展开的JSON字段
result_df = parsed_df.select("Col_A", "Col_B", "Col_C", "json_data.*")

这种方式适合JSON字段虽然动态但整体可以提前推断的场景，不需要额外关联操作。

方法2：用map转换每行生成动态Row #

直接对每行数据进行转换，把原列和JSON解析后的键值对合并成一个新Row，再重新构建DataFrame：

from pyspark.sql import Row
import json

def process_row(row):
    # 解析JSON数据
    json_data = json.loads(row.Col_C)
    # 合并原字段和JSON字段
    row_dict = row.asDict()
    row_dict.update(json_data)
    return Row(**row_dict)

# 转换所有行
rdd = df.rdd.map(process_row)
# 从RDD生成新DataFrame
result_df = spark.createDataFrame(rdd)

这种方式完全动态处理每行的JSON字段，不管每行JSON字段差异多大都能适配，缺点是需要走RDD转换，对于超大数据量可能性能稍差。

方法3：动态生成selectExpr表达式 #

如果不想用RDD，可以动态生成get_json_object的表达式，直接在DataFrame层面扩展：

from pyspark.sql import functions as F
from pyspark.sql.types import MapType, StringType

# 先获取所有可能的JSON字段（同样需要先收集）
json_fields = df.select(F.from_json(F.col("Col_C"), MapType(StringType(), StringType())).alias("json_data")) \
               .select("json_data.*").columns

# 生成select表达式：原列 + 每个JSON字段的提取表达式
select_expr = ["Col_A", "Col_B", "Col_C"] + [f"get_json_object(Col_C, '$.{field}') as {field}" for field in json_fields]
result_df = df.selectExpr(*select_expr)

这种方式纯DataFrame API操作，性能较好，适合字段数量不多的场景。

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