Airflow中的CeleryExecutor未能并行执行子DAG中的任务

Question

我们正在使用Airflow:1.10.0，并经过一些分析发现我们的一些ETL过程花费了很长时间，我们发现子DAG使用SequentialExecutor而不是使用BaseExecutor或在配置CeleryExecutor时使用。

我想知道这是Airflow的一个bug还是预期行为。在某些特定类型的任务中失去了并行执行任务的能力是没有任何意义的。

- Flavio

Answer 1

使用SequentialExecutor在subdags中是一个典型的模式，因为您通常会执行许多相似且相关的任务，并且不一定希望增加通过celery添加到队列等额外开销。有关subdags的“其他提示”部分，请参阅Airflow文档：https://airflow.apache.org/concepts.html#subdags

默认情况下，subdags设置为使用Sequential Executor（请参见：https://github.com/apache/incubator-airflow/blob/v1-10-stable/airflow/operators/subdag_operator.py#L38），但您可以更改它。

要使用celery executor，请在subdag创建中添加以下内容：

from airflow.executors.celery_executor import CeleryExecutor
mysubdag = SubDagOperator(
    executor=CeleryExecutor()
    ...
)

- Charlie Gelman

---

KubernetesExecutor的行为也是相同的吗？它是否适用于相同的情况？ - Flavio

Answer 2

也许有点晚，但实施本地执行程序对我很有效。

from airflow.executors.local_executor import LocalExecutor

subdag = SubDagOperator(
  task_id=task_id,
  default_args=default_args,
  executor= LocalExecutor(),
  dag=dag
)

- strider