Databricks Autoloader在计算集群正常，Workflow任务中无数据加载

针对你遇到的Databricks Autoloader在作业集群Workflow中无数据加载的问题，结合相同IAM账号、代码一致的前提，可从以下几个方向排查：

1. 更换Checkpoint路径为DBFS持久化路径 #

你当前使用的/tmp/_checkpoint/twitter_checkpoint是集群本地临时目录，作业集群作为临时创建的资源，本地/tmp可能存在权限限制或集群销毁时自动清理的情况，导致Autoloader无法正常写入状态数据。建议将Checkpoint路径改为DBFS上的持久化路径：

checkpoint_path = f"dbfs:/checkpoints/twitter_checkpoint"

运行前同样用dbutils.fs.rm(checkpoint_path, True)清理旧的Checkpoint数据。

2. 调整Autoloader的文件发现与触发配置 #

• 将trigger(once=True)替换为trigger(availableNow=True)：该模式会主动扫描所有可用文件并处理完成后停止，比once模式更适配作业集群的启动场景，避免因集群初始化延迟导致文件未被发现就结束任务。
• 强制使用文件列表扫描模式：添加.option("cloudFiles.useNotifications", "false")，确保Autoloader主动遍历GCS桶内文件，排除通知配置差异的影响。

修改后的核心代码片段：

query = (spark.readStream
.format("cloudFiles")
.option("cloudFiles.format", "text")
.option("cloudFiles.schemaLocation", checkpoint_path)
.option("cloudFiles.useNotifications", "false")  # 强制列表扫描
.load(file_path)
.withColumn("filePath", input_file_name())
.writeStream
.option("checkpointLocation", checkpoint_path)
.trigger(availableNow=True)  # 替换once模式
.toTable(table_name))

3. 验证作业集群对GCS桶的访问性 #

在Autoloader代码前添加文件列表验证，确认作业集群能正常读取GCS桶内的文件：

# 验证GCS桶访问
file_list = dbutils.fs.ls("gs://raw_zone_twitter")
print(f"GCS桶内文件数量: {len(file_list)}")
if len(file_list) == 0:
    raise Exception("未检测到GCS桶内的文件，请确认路径或权限")

如果输出文件数量为0，需排查作业集群的网络连通性或GCS桶的对象级权限（即使IAM账号相同，也可能存在桶的对象级权限限制）。

4. 检查作业集群的Spark配置差异 #

对比计算集群与作业集群的Spark配置，重点查看以下参数：

• spark.databricks.cloudFiles.maxFilesPerTrigger：确保未被设置为0（会阻止加载任何文件）
• spark.databricks.cloudFiles.fileListPollInterval：可适当缩短至10秒，加快文件发现速度

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