python之在 Apache Airflow 中运行超过 32 个并发任务

我正在运行 Apache Airflow 1.8.1。我想在我的实例上运行超过 32 个并发任务，但无法使任何配置正常工作。

我正在使用 CeleryExecutor，UI 中的 Airflow 配置显示 parallelism 和 dag_concurrency 为 64，我已经多次重启 Airflow 调度程序、网络服务器和工作程序(我实际上是在 Vagrant 机器上进行本地测试，但也在 EC2 实例上进行了测试)。

Airflow .cfg

# 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 = 64 
 
# The number of task instances allowed to run concurrently by the scheduler 
dag_concurrency = 64 

示例 DAG。我已经尝试过直接在 DAG 中使用和不使用 concurrency 参数。

from datetime import datetime 
 
from airflow import DAG 
from airflow.operators.bash_operator import BashOperator 
 
dag = DAG( 
    'concurrency_dev', 
    default_args={ 
        'owner': 'airflow', 
        'depends_on_past': False, 
        'start_date': datetime(2018, 1, 1), 
    }, 
    schedule_interval=None, 
    catchup=False 
) 
 
for i in range(0, 40): 
    BashOperator( 
        task_id='concurrency_dev_{i}'.format(i=i), 
        bash_command='sleep 60', 
        dag=dag 
    ) 

无论如何，同时执行的任务只有 32 个。

请您参考如下方法:

如果您有 2 个工作人员和 celeryd_concurrency = 16，那么您将被限制为 32 个任务。如果 non_pooled_task_slot_count = 32 你也会受到限制。 当然，parallelism 和 dag_concurrency 不仅需要在网络服务器和调度程序上设置为 32 以上，还需要在 worker 上设置。