无固定Schema读取时，处理Delta Lake表字段未定义的Schema匹配问题

核心思路是动态对齐现有Delta表的Schema：既不硬编码固定Schema，又能确保单小时数据缺失字段时，按表中已有Schema的类型生成null，同时支持Schema自动演进。

步骤1：获取Delta表的基准Schema #

从已有的Delta表中读取当前的Schema，作为后续数据对齐的基准，自动适配已有的Schema演进（比如之前新增过字段）。

# 读取现有Delta表的基准Schema
base_schema = spark.read.format("delta").load(path).schema

步骤2：合并推断Schema与基准Schema #

读取单小时数据时，先临时推断该批次的Schema，再和基准Schema合并：

• 基准Schema中已有的字段：保留基准的类型（比如colC的struct类型），即使当前批次没有该字段，也会保留该字段并设为null
• 当前批次新增的字段：保留推断的类型，实现Schema演进

用辅助函数实现Schema合并：

from pyspark.sql.types import StructType, StructField

def merge_schemas(base_schema: StructType, inferred_schema: StructType) -> StructType:
    # 把基准Schema字段转为字典，方便快速查找
    base_fields = {field.name: field for field in base_schema.fields}
    # 合并推断Schema的字段到基准字段中
    for field in inferred_schema.fields:
        base_fields[field.name] = field
    # 按基准Schema的原有顺序排列字段，新增字段放在末尾
    merged_fields = []
    for field in base_schema.fields:
        merged_fields.append(base_fields[field.name])
    # 添加基准Schema中没有的新增字段
    for field in inferred_schema.fields:
        if field.name not in [f.name for f in base_schema.fields]:
            merged_fields.append(field)
    return StructType(merged_fields)

步骤3：使用合并后的Schema解析JSON数据 #

替换原有的纯推断Schema，改用合并后的Schema解析Body字段，确保缺失字段（比如colC）按基准类型生成null：

# 推断当前批次的临时Schema
temp_inferred_schema = spark.read.json(df_raw.rdd.map(lambda x: x['Body'])).schema
# 合并基准Schema和临时推断Schema
final_schema = merge_schemas(base_schema, temp_inferred_schema)

# 使用合并后的Schema解析JSON
df = df_raw.withColumn('BodyNew', from_json(col('Body'), final_schema))
# 展开嵌套字段（根据业务需求选择是否执行）
df = df.selectExpr("hour", "BodyNew.*")

步骤4：写入Delta表时保留Schema演进配置 #

保持原有的写入配置，mergeSchema=true确保新增字段能被合并到Delta表中：

opts = {
    "mergeSchema": "true",
    "overwriteSchema": "false",
    "partitionOverwriteMode": "dynamic"
}

df.write.partitionBy("hour")\
    .mode("overwrite")\
    .format("delta")\
    .options(**opts)\
    .save(path)

关键逻辑说明 #

• 当第3小时数据中colC未定义时，合并后的Schema会保留colC的struct类型，from_json自动为该字段填充null，与Delta表现有类型完全匹配，避免Schema不匹配错误
• 如果后续新数据中colC新增字段（比如field3），合并后的Schema会包含这个新字段，写入时mergeSchema=true会自动将该字段添加到Delta表的Schema中，实现无感知的Schema演进
• 基准Schema动态从Delta表读取，无需硬编码，完全适配未来的Schema变化

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