在已有Airflow环境中用Celery和Redis运行自定义任务遇超时求助

你的代码出现TimeoutError的核心原因是：你自定义的Celery应用和Airflow自带的Celery实例完全独立。Airflow的Celery Workers只会处理Airflow调度的DAG任务，不会执行你自己定义的maximum这类自定义Celery任务，导致任务一直处于等待状态，最终超时。

以下是几种可行的解决思路：

1. 复用Airflow的Celery实例（推荐） #

Airflow已经初始化了自己的Celery应用，你可以直接导入使用，无需重新创建新的Celery app，这样自定义任务会由Airflow的现有Worker执行：

from airflow.executors.celery_executor import app as airflow_celery_app

@airflow_celery_app.task
def maximum(x, y):
    print("here")
    print(x)
    return x if x > y else y

def test():
    max1 = maximum.delay(5, 4)
    # 适当延长超时时间，避免因Airflow Worker繁忙导致的延迟
    print(max1.get(timeout=10))
    return 0

⚠️ 注意：这种方式会占用Airflow的Worker资源，如果自定义任务资源消耗较大，可能影响Airflow本身的DAG调度效率，需根据实际场景评估。

2. 启动独立的Celery Worker处理自定义任务 #

如果你希望隔离Airflow任务和自定义任务的资源，可以启动专门的Worker来处理你的自定义Celery应用：

1. 确保你的自定义Celery app代码（包含app = Celery('main', broker=..., backend=...)和maximum任务定义）能被Worker访问到；
2. 在容器中执行启动命令：

   celery -A main worker --loglevel=info
   

   （这里的main是你定义Celery app的Python模块名）

启动后，你的自定义任务会被这些独立的Worker执行，不会干扰Airflow的正常运行。

3. 改用Airflow原生组件实现并行（更贴合Airflow设计） #

如果你的需求只是实现简单的并行计算，完全可以利用Airflow自身的调度能力，无需手动调用Celery：

比如将计算逻辑拆分为独立的Airflow任务，通过CeleryExecutor实现并行：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def compute_max(x, y, **context):
    result = x if x > y else y
    context['ti'].xcom_push(key='max_result', value=result)

def get_result(**context):
    max_result = context['ti'].xcom_pull(key='max_result')
    print(max_result)

with DAG(
    'parallel_compute_dag',
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
) as dag:
    compute_task = PythonOperator(
        task_id='compute_max',
        python_callable=compute_max,
        op_kwargs={'x':5, 'y':4},
        provide_context=True
    )

    result_task = PythonOperator(
        task_id='get_result',
        python_callable=get_result,
        provide_context=True
    )

    compute_task >> result_task

这种方式完全遵循Airflow的设计模式，无需额外维护Celery实例，更适合Airflow场景下的任务并行需求。

内容的提问来源于stack exchange，提问作者atul.mishra