关于CeleryExecutor无法执行多分支Pipeline的问题咨询

问题概述 #

我们发现CeleryExecutor无法执行多分支Pipeline，而LocalExecutor可正常执行该多分支任务。此外，CeleryExecutor会在Redis中为一个DAG创建一条消息，请问这是预期行为吗？

当前CeleryExecutor仅会选取taskA1或taskB1其中一条执行流，另一条则始终处于排队状态。

任务流程示例 #

多分支任务依赖 #

taskA1 -> taskA2 -> taskA3
|
taskB1 -> taskB2 --|

其他依赖示例 #

• 线性依赖：

  a >> b
  b >> c
  

• 独立分支依赖：

  a1 >> b1
  b1 >> c1
  

分析与解答 #

多分支任务排队问题 #

这种表现不是CeleryExecutor的预期行为，通常由以下配置或机制问题导致：

1. Worker并发数限制
   检查Airflow配置中的celeryd_concurrency参数，若该值设置为1，单个Celery Worker一次只能处理一个任务，会直接导致另一条分支的任务处于排队状态。需根据服务器资源情况调高该值，确保Worker具备并行处理多任务的能力。

2. 队列分配与消费策略
   若两条分支的任务被分配至同一Celery队列，且该队列对应的Worker并发不足，会引发任务排队。建议为不同分支的任务指定独立队列，并配置专属Worker消费对应队列，实现任务分流并行。

3. DAG任务依赖配置
   确认DAG中taskA1和taskB1的依赖配置是否正确，确保二者均为无上游依赖的起始任务；同时检查DAG的schedule_interval、max_active_runs等参数，确认未限制任务的并行执行。

Redis中单个DAG对应一条消息的机制 #

这是CeleryExecutor的正常预期行为：Airflow会将DAG的调度指令封装为单条消息发送至Redis（Celery的消息代理），Worker从Redis中拉取任务消息后，会解析DAG内的所有可执行任务并分配执行。单条消息不影响多分支任务的并行性，只要Worker有足够并发能力，即可同时处理多条分支的任务。

内容的提问来源于stack exchange，提问作者Sanketha C R