Databricks AutoLoader处理大量空Parquet文件异常排查求助

1. 异步流未等待完成：在wheel脚本、循环调用Notebook场景中，流查询通过start()启动后是异步执行的，若未显式调用awaitTermination()等待处理完成，进程会直接退出，导致流处理被强制终止，无法生成Delta文件。而单独运行Notebook时，Databricks会自动维持会话直到流任务结束，因此能正常处理。
2. 空文件无数据输出：空Parquet文件读取后无有效数据，append输出模式下不会生成任何Delta文件；若此时进程提前退出，目标目录会呈现为空状态。
3. 会话生命周期限制：循环调用Notebook时，被调用的Notebook执行结束后会话会关闭，后台运行的流查询会被Databricks强制终止，无法完成全量文件处理。

1. 强制等待流查询完成 #

在所有启动流查询的代码中，必须添加awaitTermination()，确保流处理完所有文件后再退出进程：

# 启动流写入
query = df.writeStream.format("delta")
  .outputMode("append")
  .option("checkpointLocation", <path_to_checkpoint>)
  .queryName(<processed_table_name>)
  .partitionBy(<partition-key>)
  .option("mergeSchema", True)
  .trigger(once=True)
  .start(<path-to-target>)

# 等待流处理完成，直到查询自然结束
query.awaitTermination()

2. 显式处理空文件场景 #

若需要即使无数据也生成空Delta表（避免目标目录为空），可在读取后判断记录数，为空则主动创建空表：

# 读取流数据
df = spark.readStream.format("cloudFiles")
     .options(**CLOUDFILE_CONFIG)
     .option("recursiveFileLookup", True)
     .schema(schema)
     .option("locale", "de-DE")
     .option("dateFormat", "dd.MM.yyyy")
     .option("timestampFormat", "MM/dd/yyyy HH:mm:ss")
     .load(<path-to-source>)

# 统计读取到的记录数（仅用于判断，不影响流处理逻辑）
record_count = df.groupBy().count().collect()[0][0]

if record_count == 0:
    # 创建空Delta表
    spark.createDataFrame([], schema=schema).write.format("delta").mode("append").save(<path-to-target>)
else:
    # 正常执行流写入并等待完成
    query = df.writeStream...start()
    query.awaitTermination()

3. 优化AutoLoader配置 #

• 合并重复配置：将cloudFiles.format、pathGlobFilter统一放入CLOUDFILE_CONFIG，避免重复设置；
• 针对大量小文件，关闭通知模式改用列表批量模式（处理现有文件更高效）：

  CLOUDFILE_CONFIG.update({
      "cloudFiles.useNotifications": False,
      "cloudFiles.useListBatch": True
  })
  

4. 修正循环调用Notebook逻辑 #

确保被调用的Notebook中包含awaitTermination()，并为dbutils.notebook.run()设置足够长的超时时间，避免因超时提前终止任务：

for item in active_tables_metadata:
    # 设置超时时间为3600秒（可根据实际处理时长调整）
    dbutils.notebook.run("process_raw", 3600, item)

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