Databricks Autoloader全量加载后Bronze层表为空问题求助

核心问题点 #

• 路径不匹配：你获取Schema的路径是/mnt/landing/20240826/SalesLT.Address.parquet，但Autoloader加载的路径是/mnt/landing/SalesLT.Address.parquet，如果实际文件存放在日期子目录下，Autoloader会找不到目标文件，自然没有数据写入。
• Checkpoint目录冲突：将Autoloader的cloudFiles.schemaLocation和流写入的checkpointLocation设为同一目录，两类元数据相互干扰，导致Autoloader无法正确识别文件或处理状态。
• 输出模式配置矛盾：同时设置outputMode("append")和.option("mode", "overwrite")，这两个配置逻辑冲突——outputMode定义流处理的输出规则，而Delta的覆盖写入需要用.mode("overwrite")（不是option参数），且全量覆盖场景更适配outputMode("complete")。
• Checkpoint残留状态：如果之前运行过任务，即使清理了落地目录，checkpoint中记录的已处理文件标记会让Autoloader跳过新放入的文件，导致无数据写入。

修正方案 #

1. 修正文件加载路径 #

确保Autoloader指向包含目标文件的目录，而非单个文件（Autoloader支持读取目录下所有符合格式的文件）：

# 示例：指向日期子目录或根落地目录
load_path = "/mnt/landing/20240826/"

2. 分离Schema与Checkpoint目录 #

为两类元数据设置独立目录，避免冲突：

schema_loc = "/mnt/landing/_schema/address_autoload/"
checkpoint_loc = "/mnt/landing/_checkpoint/address_autoload/"

3. 正确配置全量覆盖逻辑 #

针对每日全量场景，推荐两种实现方式：

方式一：流处理（Trigger Once模式）

使用outputMode("complete")配合.mode("overwrite")实现全量覆盖：

(spark.readStream
 .format("cloudFiles")
 .option("cloudFiles.format", "parquet")
 .option("cloudFiles.schemaLocation", schema_loc)
 .schema(ddl)
 .load(load_path)
 .writeStream
 .format("delta")
 .outputMode("complete")
 .mode("overwrite")
 .option("checkpointLocation", checkpoint_loc)
 .trigger(once=True)
 .toTable("adventureworks.address"))

方式二：批处理（更适配全量场景）

因为是每日全量文件且会清理目录，批处理更简单直接，规避流处理的checkpoint问题：

# 读取目录下所有Parquet文件
df = spark.read.parquet(load_path)
# 覆盖写入Delta表
df.write.format("delta").mode("overwrite").saveAsTable("adventureworks.address")

4. 清理残留状态 #

若之前运行过任务，需删除旧的schema和checkpoint目录，确保Autoloader重新识别新文件：

dbutils.fs.rm(schema_loc, recurse=True)
dbutils.fs.rm(checkpoint_loc, recurse=True)

5. 优化Schema验证逻辑 #

自动判断Bronze层表是否存在，动态获取Schema：

# 检查目标表是否存在
table_exists = spark.catalog.tableExists("adventureworks.address")

if table_exists:
    # 从已有表获取Schema
    schema = spark.table("adventureworks.address").schema
    ddl = schema.toDDL()
else:
    # 从落地区文件获取Schema
    sample_df = spark.read.parquet("/mnt/landing/20240826/SalesLT.Address.parquet")
    ddl = sample_df.schema.toDDL()

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