如何合并多个不同schema为单一数据集以支持后续动态pivot操作

• 第一步先建立全局字段映射配置表，无需把规则写死在处理代码里。配置表仅需保留3个核心字段：原始数据源标识、原始列名、统一标准列名，所有同含义的异名列都提前录入配置即可，后续新增数据源只需要更新配置表，无需修改主处理逻辑。
• 第二步动态生成每个数据源的查询逻辑，无需逐个手动编写代码。遍历所有输入数据源，读取每个数据源的原生schema后和映射配置表匹配：匹配到规则的字段直接重命名为统一列名，同时可以同步做类型对齐；没有匹配到规则的非必要字段直接忽略，或按需求填充null值；建议额外新增source_table固定字段标记数据来源，方便后续问题排查。
  单个数据源的动态查询逻辑示例（自动生成无需手写）：

SELECT 
  原始列1 AS 统一列名A,
  原始列2 AS 统一列名B,
  CAST(原始列3 AS STRING) AS 统一列名C, -- 同步做类型统一，避免合并时报类型不兼容错误
  '数据源1' AS source_table
FROM 数据源1

• 第三步把所有数据源的查询逻辑用UNION ALL直接拼接，一步生成统一schema的数据集，全程不需要创建任何中间表。整个处理链路完全动态，新增数据源仅需更新映射配置，不需要修改主代码、也不需要新增中间数据集。

• 映射配置表可新增「标准字段类型」字段，生成查询逻辑时自动做CAST转换，避免不同数据源同含义字段类型差异导致合并报错
• 可增加前置校验逻辑：每次生成处理逻辑时自动检查所有统一列的映射规则是否完整，缺映射的直接提前抛错，不会把问题带到后续的pivot步骤
• 如果使用Spark/PySpark这类分布式处理引擎，可直接用动态列重命名+unionByName的逻辑实现，代码更简洁：

import functools
from pyspark.sql.functions import lit

# 映射规则可从配置表读取，不用写死在代码里
col_mapping = {
    "数据源1": {"old_col1": "standard_col1", "old_col2": "standard_col2"},
    "数据源2": {"old_col_a": "standard_col1", "old_col_b": "standard_col2"}
}
standard_cols = ["standard_col1", "standard_col2", "source_table"]

df_list = []
for source_name, map_rule in col_mapping.items():
    df = spark.read.table(source_name)
    # 按规则重命名列
    for old_col, new_col in map_rule.items():
        df = df.withColumnRenamed(old_col, new_col)
    # 补全所有标准列，不存在的填null
    for col in standard_cols:
        if col not in df.columns:
            df = df.withColumn(col, lit(None))
    # 仅保留标准列，避免多余字段干扰
    df = df.select(*standard_cols)
    df_list.append(df)

# 合并所有数据集直接得到统一schema结果
final_unified_df = functools.reduce(lambda a,b: a.unionByName(b), df_list)

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