celery - 如果传感器数量大于并发数，Airflow celery Worker 会被阻塞吗？

Question

假设我将 celery 并发设置为 n，但我有 m(m>n)一个dag中的ExternalTask​​Sensor，它会检查另一个名为do_sth的dag，这些ExternalTask​​Sensor将消耗所有celeryworker，因此实际上没有人会工作。

但我不能将并发设置得太高(例如 2*m)，因为 dag do_sth 可能会启动太多进程，从而导致内存不足。

我很困惑应该为 celery 并发设置多少数字？

Answer 1

在 ETL best practices with Airflow's Gotchas section作者解决了这个普遍问题。建议之一是为您的传感器任务设置一个池，这样您的其他任务就不会被饿死。根据您的情况，确定您希望一次运行的传感器任务数量(小于并发级别)，并以此为限制设置一个池。设置池后，将池参数传递给每个传感器运算符(operator)。有关池的更多信息，请参阅 Airflow's documentation on concepts 。以下是将池参数传递给运算符的示例:

aggregate_db_message_job = BashOperator( 
    task_id='aggregate_db_message_job', 
    execution_timeout=timedelta(hours=3), 
    pool='ep_data_pipeline_db_msg_agg',
    bash_command=aggregate_db_message_job_cmd, dag=dag)