airflow - 在 Apache Airflow 中运行并行任务

Question

我能够配置 airflow.cfg 文件来依次运行任务。

我想做的是并行执行任务，例如一次 2 个并到达列表末尾。

我该如何配置？

Answer 1

在 Airflow 中并行执行任务取决于您使用的执行器，例如 SequentialExecutor、LocalExecutor、CeleryExecutor 等。

为了进行简单的设置，您只需在airflow.cfg中将执行器设置为LocalExecutor即可实现并行性:

[core]
executor = LocalExecutor

引用:https://github.com/apache/incubator-airflow/blob/29ae02a070132543ac92706d74d9a5dc676053d9/airflow/config_templates/default_airflow.cfg#L76

这将为每个任务启动一个单独的进程。

(当然，您需要有一个包含至少 2 个可以并行执行的任务的 DAG 才能看到其工作原理。)

或者，使用 CeleryExecutor，您可以通过运行(任意次数)来启动任意数量的工作线程:

$ airflow worker

任务将进入 Celery 队列，每个 Celery 工作线程将从队列中退出。

您可能会发现 Airflow 配置文档中的使用 Celery 进行扩展部分很有帮助。

https://airflow.apache.org/howto/executor/use-celery.html

对于任何执行器，一旦运行，您可能需要调整控制并行性的核心设置。

它们都可以在[core]下找到。这些是默认值:

# The amount of parallelism as a setting to the executor. This defines
# the max number of task instances that should run simultaneously
# on this airflow installation
parallelism = 32

# The number of task instances allowed to run concurrently by the scheduler
dag_concurrency = 16

# Are DAGs paused by default at creation
dags_are_paused_at_creation = True

# When not using pools, tasks are run in the "default pool",
# whose size is guided by this config element
non_pooled_task_slot_count = 128

# The maximum number of active DAG runs per DAG
max_active_runs_per_dag = 16

引用:https://github.com/apache/incubator-airflow/blob/29ae02a070132543ac92706d74d9a5dc676053d9/airflow/config_templates/default_airflow.cfg#L99