如何在Databricks中自动刷新含新增列的时序DataFrame

针对你在Databricks 10.4 LTS中遇到的CSV文件列新增导致读取报错的问题，以下是几种可行的流水线内自动处理方案：

方案1：直接读取整个目录（最简洁高效） #

Spark的CSV读取器支持mergeSchema选项，可自动合并目录下所有文件的Schema，缺失列自动填充为Null。若所有文件存储在同一父目录下，优先使用此方法：

final_df = spark.read.format("csv") \
    .option("sep", "\t") \
    .option("encoding", "utf-16") \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .option("mergeSchema", "true") \
    .load("/path/to/your/csv/files")

核心逻辑： #

• mergeSchema=true会扫描目录下所有文件的表头，自动合并出包含所有列的完整Schema
• 读取时，对无对应列的文件自动将该列值设为Null
• 无需手动遍历文件，Spark并行处理性能更优

方案2：手动合并Schema后逐个读取（适配分散文件场景） #

如果文件分散在不同路径无法批量读取，可先收集所有文件的完整Schema，再用统一Schema读取每个文件：

步骤1：生成完整Schema #

from pyspark.sql.types import StructType

# 收集所有文件路径
dfdir = spark.sql("SELECT path FROM your_file_path_table")
file_paths = [row.path for row in dfdir.collect()]

# 遍历所有文件，合并出包含所有列的Schema
full_schema = StructType()
for path in file_paths:
    # 仅读取表头获取Schema，不加载全量数据
    temp_schema = spark.read.format("csv") \
        .option("sep", "\t") \
        .option("encoding", "utf-16") \
        .option("header", "true") \
        .option("inferSchema", "true") \
        .load(path).schema
    
    # 添加当前文件有但完整Schema中缺失的列
    for field in temp_schema.fields:
        if field.name not in full_schema.names:
            full_schema.add(field)

步骤2：用统一Schema读取并合并文件 #

final_df = None
for path in file_paths:
    # 用完整Schema读取单个文件
    single_df = spark.read.format("csv") \
        .option("sep", "\t") \
        .option("encoding", "utf-16") \
        .option("header", "true") \
        .schema(full_schema) \
        .load(path)
    
    # 合并到最终DataFrame，允许缺失列（自动填充Null）
    if final_df is None:
        final_df = single_df
    else:
        final_df = final_df.unionByName(single_df, allowMissingColumns=True)

方案3：基于最新文件Schema读取（性能优化） #

如果确定最新文件包含所有新增列，可直接读取最新文件的Schema，避免遍历所有文件：

# 按文件名（假设含日期标识）排序取最新文件
latest_file = sorted(file_paths, key=lambda x: x.split("/")[-1])[-1]

# 获取最新文件的完整Schema
full_schema = spark.read.format("csv") \
    .option("sep", "\t") \
    .option("encoding", "utf-16") \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .load(latest_file).schema

# 后续读取合并逻辑同方案2的步骤2

关键注意事项： #

• 若需固定列数据类型，可手动定义Schema替换inferSchema=true的自动推断结果
• unionByName(allowMissingColumns=True)确保合并时自动对齐列名，避免列顺序不一致导致的报错

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