使用Databricks Autoloader时如何验证列名与顺序后写入Delta表？

一、动态验证传入文件的Schema（列名+顺序）并拒绝不匹配文件 #

你可以通过以下两种方式实现严格的Schema验证：

1. 基于Autoloader内置约束+自定义顺序检查 #

先获取目标Delta表的完整Schema（包含列顺序），结合Autoloader的Schema限制参数，再添加顺序校验逻辑：

# 获取目标Delta表的Schema（保留列顺序、类型信息）
target_schema = spark.read.table("my_table").schema
# 提取目标列名的顺序列表
target_columns = [field.name for field in target_schema.fields]

query = (
    spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "parquet")
    # 强制使用目标Schema读取文件，缺失列/类型不匹配直接报错
    .schema(target_schema)
    # 禁止Schema演化，遇到新列直接终止读取
    .option("cloudFiles.schemaEvolutionMode", "failOnNewColumns")
    .load("path")
    # 强制列顺序与目标表一致
    .select(*target_columns)
    .writeStream
    .format("delta")
    .outputMode("append")
    # 批次级最终验证，不匹配则抛出异常终止写入
    .foreachBatch(lambda batch_df, batch_id: 
        if batch_df.schema != target_schema:
            raise ValueError(f"批次 {batch_id} Schema与目标表不匹配")
        else:
            batch_df.write.format("delta").mode("append").saveAsTable("my_table")
    )
    .option("checkpointLocation", "/path/to/checkpoint")
    .start()
)

• schema(target_schema)强制Autoloader按目标Schema解析文件，类型/列缺失会直接触发错误；
• select(*target_columns)确保写入列的顺序完全对齐目标表；
• foreachBatch的额外校验可以捕获隐性不匹配（如自动类型转换导致的差异），不匹配批次不会写入，Autoloader也不会标记对应文件为已处理。

2. 逐文件Schema验证 #

如果需要对单个文件做精准校验，可以开启文件元数据获取，逐文件对比Schema：

target_schema = spark.read.table("my_table").schema

def validate_single_file(file_path):
    # 读取单个Parquet文件的完整Schema
    file_schema = spark.read.parquet(file_path).schema
    # 对比Schema的列名、类型、顺序是否完全一致
    return file_schema == target_schema

query = (
    spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "parquet")
    # 开启文件元数据，获取每个数据对应的源文件路径
    .option("cloudFiles.includeFileMetadata", "true")
    .load("path")
    # 过滤掉Schema不匹配的文件数据
    .filter(lambda row: validate_single_file(row["_metadata"]["file_path"]))
    .select(*[field.name for field in target_schema.fields])
    .writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/path/to/checkpoint")
    .toTable("my_table")
)

• cloudFiles.includeFileMetadata会在数据中添加_metadata字段，包含源文件路径；
• 自定义校验函数逐文件验证，不匹配的文件数据会被直接过滤，不会写入Delta表。

二、让Autoloader检查文件头但不推进索引 #

要实现“扫描文件Schema但不标记为已处理”的需求，可通过以下两种方式：

1. 批处理模式预览Schema #

用Autoloader的批处理读取方式扫描文件，不会写入checkpoint，因此不会推进索引：

target_schema = spark.read.table("my_table").schema

# 扫描待处理文件的Schema，不加载数据，不更新索引
preview_schema = (
    spark.read
    .format("cloudFiles")
    .option("cloudFiles.format", "parquet")
    .option("cloudFiles.schemaLocation", "/path/to/schema_location")
    # 仅扫描文件Schema，不加载数据内容
    .option("cloudFiles.maxFilesPerTrigger", 0)
    .load("path")
    .schema
)

# 对比Schema是否匹配
if preview_schema != target_schema:
    print("待处理文件存在Schema不匹配情况")

• cloudFiles.maxFilesPerTrigger=0会让Autoloader扫描所有待处理文件，但仅提取Schema不加载数据；
• 该操作不会写入checkpoint，后续流处理仍会处理这些文件。

2. 临时Checkpoint扫描 #

创建临时checkpoint目录执行单次扫描，完成后删除临时目录，不影响原流的索引状态：

temp_checkpoint = "/path/to/temp_checkpoint"
target_schema = spark.read.table("my_table").schema

# 临时启动流扫描文件Schema
preview_query = (
    spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "parquet")
    .option("cloudFiles.schemaLocation", "/path/to/schema_location")
    .load("path")
    # 不实际写入数据
    .writeStream
    .format("noop")
    .option("checkpointLocation", temp_checkpoint)
    # 仅执行一次扫描
    .trigger(once=True)
    .start()
)

preview_query.awaitTermination()

# 获取扫描到的文件Schema
scanned_schema = spark.read.parquet("path").schema

# 删除临时checkpoint，避免影响原流
dbutils.fs.rm(temp_checkpoint, recurse=True)

• 使用noop写入器不会产生任何数据输出；
• trigger(once=True)仅执行一次扫描，完成后删除临时checkpoint，原流的索引不会被修改。

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