Airflow技术问题：如何让单个Worker执行某一DAG的所有任务？

刚好碰到过一模一样的场景！多Worker下跨节点的文件依赖问题确实挺闹心的，既然你已经在使用Queue功能，其实可以基于现有配置做几个简单调整，就能实现指定DAG仅由单个Worker执行的需求，下面给你几个实用方案：

方案一：给目标DAG绑定专属Queue，仅让单个Worker监听该Queue #

这是最直接也最容易维护的方法，完全兼容你现有的Queue配置：

• 第一步：给需要单Worker执行的DAG指定专属Queue。你可以在DAG代码里配置：

  with DAG(
      dag_id="your_file_processing_dag",
      # 其他DAG参数
      default_args={
          "queue": "single_worker_file_queue",  # 专属队列名称
          # 其他默认参数
      },
  ) as dag:
      # 任务定义...
  

• 第二步：启动单个Celery Worker，让它只监听这个专属Queue（如果需要它也处理默认队列的任务，可以把default也加上）：

  celery worker -A airflow.executors.celery_executor.app -Q single_worker_file_queue
  

• 原理：这个DAG的所有任务都会被Airflow调度到专属Queue里，而只有你指定的这个Worker会消费该Queue的任务，自然就能保证所有文件下载、处理、上传的任务都在同一个Worker节点上执行，不会出现文件找不到的问题。

方案二：给专属Worker设置单并发（可选，适合无并行需求的DAG） #

如果你的文件处理任务本身不需要并行执行，还可以给这个专属Worker设置并发数为1，进一步避免同一Worker内的任务资源冲突：

celery worker -A airflow.executors.celery_executor.app -Q single_worker_file_queue --concurrency=1

这样这个Worker同一时间只会处理一个任务，也能规避比如文件读写锁这类额外的问题。

方案三：利用Celery任务亲和性（细粒度控制场景） #

如果不想单独创建Queue，也可以通过Celery的任务亲和性配置，让同一个DAG的所有任务都分配到同一个Worker上：
在DAG的default_args里添加Celery相关配置：

default_args = {
    # 其他默认参数...
    'queue': 'your_existing_queue',
    'celery_kwargs': {
        'routing_key': f"dag_{{{{ dag.dag_id }}}}",
        'queue': 'your_existing_queue'
    }
}

然后启动Worker时指定只接收对应路由键的任务：

celery worker -A airflow.executors.celery_executor.app -Q your_existing_queue -X routing_key=dag_your_file_processing_dag

不过这个方法相对复杂，不如专属Queue的方式直观，适合有特殊细粒度需求的场景。

小提醒：如果你已经在用Queue做其他任务的路由，这个方案完全不会影响现有配置，只是给需要单Worker的DAG单独分配了一个专属的Queue节点而已。

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