如何用DataFrame存储的SparkSQL表达式实现多Schema JSON字段归一化

最优实现方案（性能最高，无额外Shuffle） #

你之前用CASE语句的方案本身性能非常好，Spark对固定逻辑的CASE语句优化非常成熟，仅需将手动拼接SQL的过程改为从expressions_df自动生成即可，不需要引入关联操作避免行膨胀和Shuffle开销。
实现代码如下：

from pyspark.sql import functions as F

# 1. 从表达式元数据表提取映射规则（小表直接拉取到Driver端无压力）
expr_rows = expressions_df.collect()
# 按目标标准化列分组，存储每个列对应不同schema的取值表达式
col_rule_map = {}
for row in expr_rows:
    col = row["column"]
    schema = row["schema"]
    expr = row["column_expression"]
    if col not in col_rule_map:
        col_rule_map[col] = []
    col_rule_map[col].append((schema, expr))

# 2. 自动生成每个目标列的CASE表达式
select_list = []
for target_col, rules in col_rule_map.items():
    case_when_part = " ".join([f"WHEN schema = '{s}' THEN {e}" for s, e in rules])
    full_expr = f"CASE {case_when_part} END AS {target_col}"
    select_list.append(F.expr(full_expr))

# 3. 直接对原始数据表执行查询得到结果
result_df = raw_df.select(*select_list)

该方案的优势：

• 无关联、无Shuffle，性能和你手动写的SQL完全一致，即使扩展到上百个Schema、几十列也不会有明显性能下降
• 完全动态生成，新增Schema或标准化列不需要修改业务代码

关联+Pivot实现方案（仅做参考，性能低于前者） #

如果一定要走关联表达式表的逻辑，可以用行转宽的方式实现，但会产生Shuffle开销，适合表达式规则量极大、生成CASE语句过长的极端场景：

result_df = (
    raw_df
    # 给原始数据每行加唯一标识，避免重复行聚合出错
    .withColumn("row_unique_id", F.monotonically_increasing_id())
    .join(expressions_df, on="schema", how="inner")
    # 执行表达式得到对应列的值
    .withColumn("col_value", F.expr("column_expression"))
    # 按行分组转宽，得到标准化列结构
    .groupBy("row_unique_id")
    .pivot("column")
    .agg(F.first("col_value"))
    .drop("row_unique_id")
)

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