Airflow多Celery Worker场景下依赖文件的DAG执行文件缺失如何解决

Airflow 使用 CeleryExecutor 时，所有任务会被独立调度到空闲的 Worker 节点执行。单 Worker 场景下，下载、处理两个任务都运行在同一台节点，保存在本地的文件可以共享；扩容到多 Worker 后，下载任务可能调度到 Worker A 把文件存在 A 的本地磁盘，处理任务被调度到 Worker B，B 本地自然找不到对应的文件。

下面对应你的三个疑问逐一给出方案：

1. 使用 SFTPHook 合并下载、处理任务（最推荐） #

这个方案完全可行，也是同类场景的最优解。把原本拆分的SFTPOperator和PythonOperator合并为一个 Python 任务，在同一个任务上下文里先完成文件下载，再执行处理逻辑，整个流程只会被调度到同一个 Worker 上，不存在跨节点文件访问问题。
参考代码示例：

import os
from airflow.decorators import task
from airflow.providers.sftp.hooks.sftp import SFTPHook

@task
def download_and_process_file(sftp_conn_id: str, remote_path: str, temp_local_path: str):
    # 初始化SFTPHook，复用你的SFTP连接配置
    sftp_hook = SFTPHook(sftp_conn_id=sftp_conn_id)
    # 下载文件到当前Worker本地临时路径
    sftp_hook.retrieve_file(remote_full_path=remote_path, local_full_path=temp_local_path)
    
    # 直接在当前任务内处理文件，无需跨任务传递本地路径
    with open(temp_local_path, 'r', encoding='utf-8') as f:
        # 此处写你的文件处理逻辑
        process_result = [line.strip() for line in f.readlines()]
    
    # 处理完成后删除本地临时文件，避免占用磁盘空间
    os.remove(temp_local_path)
    return process_result

这个方案不需要修改集群配置，也不会降低分布式调度的灵活性，适配绝大多数场景。

2. 限制关联任务调度到同一Worker #

可以实现，利用 Celery 的队列机制就能做到：

• 启动指定 Worker 时，给它绑定专属队列，执行命令：airflow celery worker -q sftp_process_queue
• 给你的下载、处理两个任务都添加参数queue='sftp_process_queue'，指定这两个任务只会被分配到监听了sftp_process_queue的 Worker 上
  这个方案的缺点是降低了调度灵活性，如果绑定的专属 Worker 故障，这两个任务会直接阻塞，只适合临时低可用要求的场景使用。

3. 共享存储方案 #

如果必须拆分两个任务且不想限制调度，可以把下载的文件存储到所有 Worker 都能访问的共享存储（比如NFS、分布式文件系统、对象存储），下载任务将文件写入共享存储，处理任务从共享存储读取文件，不管调度到哪个 Worker 都能正常访问。

优先选合并任务的方案，代码改动量最小，没有额外依赖；如果业务逻辑必须拆分两个任务，优先用共享存储方案；只有特殊场景下再考虑限制调度队列的方案。

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