Airflow DAG 在不同工作节点上执行并行任务

Question

我有一个由 3 个工作节点组成的 Airflow 集群，其中 CeleryExecutor 和 RabbitMQ 用于通信。 我的 DAG 通常由下载文件、解压缩文件、上传到 hadoop 等任务组成。因此它们相互依赖并且必须在单个机器/节点上运行。

当气流调度单个 DAG 的这些任务时，在不同的节点上，我最终会出错，因为这些任务被调度在不同的机器上，但我需要将 DAG 中的所有任务调度在一个单机。

我尝试在气流.cfg 和初始化 dag 时设置 dag_concurrency = 1 和 max_active_runs_per_dag = 1，但没有成功。

我的气流.cfg 的其余部分：

parallelism = 32
dag_concurrency = 1
worker_concurrency = 16
max_active_runs_per_dag = 16

据我了解，将 dag_concurrency 设置为 1 应该可以解决问题，但我在这里缺少什么？

Answer 1

CeleryExecutor 支持多个queues，您可以为每个操作员定义一个特定的队列（是BaseOperator 的一个属性），然后为每个工作人员订阅该特定队列。请注意，worker 可以监听一个或多个队列。

来自docs：

Worker 可以监听一个或多个任务队列。当工人 启动（使用命令airflow celery worker），一组 可以指定逗号分隔的队列名称（例如，airflow celery 工人 -q 火花）。然后，该工作人员将仅拾取连接到的任务 指定的队列

这是一个 DAG 示例：

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.utils.dates import days_ago


default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': days_ago(1),
}
dag = DAG('dist_example',
          schedule_interval='@once',
          catchup=False,
          default_args=default_args
          )
get_hostname = 'echo $(hostname)'

t1 = BashOperator(
    task_id='task_for_q1',
    bash_command=get_hostname,
    queue='queue_1',
    dag=dag
)
t2 = BashOperator(
    task_id='task_for_q2',
    bash_command=get_hostname,
    queue='queue_2',
    dag=dag
)
t1 >> t2

worker_1： airflow celery worker -q default,queue_1

worker_2： airflow celery worker -q default,queue_2

通过侦听您的特定队列和default（由default_queue 配置键定义），您不会影响任何其他任务的标准多工人行为。

希望对你有用！