Airflow 2.6：通过运行时配置设置手动触发DAG的任务超时

问题背景 #

我有一个需通过UI手动触发的DAG，可传入参数指定数据子集，不同子集对应的任务运行时长从5分钟到1小时不等。希望通过UI的运行时配置传入任务超时时间（execution_timeout），而非设置上限以适配最大子集。

尝试过的实现方式：

with DAG(
    ...
    params={"mls_sid" : "0", "time_out" : 10}
) as dag:
    task = PythonOperator(
        ...
        execution_timeout = timedelta(minutes="{{ params.time_out }}")
)

遇到的问题：直接传入字符串给timedelta会报错，用int()包裹Jinja模板则出现十进制值错误，怀疑DAG加载时未使用定义的默认值。想知道是否无法为execution_timeout使用模板，或者有没有其他运行时（UI、API或其他DAG触发时）传入execution_timeout的方式。

解决方案 #

首先明确：Airflow的execution_timeout参数不支持Jinja模板渲染。因为这个参数是在DAG解析阶段（而非任务运行阶段）被评估的，此时运行时参数还未注入，模板无法生效。

以下是三种可行的替代方案：

方案1：任务内部手动实现超时控制（推荐） #

放弃使用Airflow原生的execution_timeout，在任务代码中通过Python的signal模块自定义超时逻辑，直接读取运行时传入的params参数。

示例代码：

import signal
from airflow.decorators import dag, task
from datetime import datetime, timedelta

def timeout_handler(signum, frame):
    raise TimeoutError("任务运行超时")

@dag(
    start_date=datetime(2024, 1, 1),
    params={"mls_sid": "0", "time_out": 10},
    schedule=None
)
def dynamic_timeout_dag():
    @task
    def process_data_subset(params):
        # 从运行时参数中读取超时时间并设置信号
        timeout_minutes = int(params["time_out"])
        signal.signal(signal.SIGALRM, timeout_handler)
        signal.alarm(timeout_minutes * 60)
        
        # 这里编写你的核心任务逻辑
        print(f"开始处理数据子集 {params['mls_sid']}，超时设置为 {timeout_minutes} 分钟")
        
        # 任务正常完成后取消超时信号
        signal.alarm(0)
    
    process_data_subset()

dag = dynamic_timeout_dag()

注意：该方案依赖signal模块，不适用于Windows环境，部分容器化部署场景可能需要调整信号权限。

方案2：通过Airflow Variable传递超时参数 #

通过前置任务将触发时的超时参数写入Airflow Variable，主任务读取该Variable并控制内部逻辑，同时设置一个足够大的默认execution_timeout作为兜底。

示例代码：

from airflow import DAG, Variable
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

def set_timeout_var(**context):
    timeout = context["params"]["time_out"]
    # 用DAG Run ID作为Variable键，避免并发触发时的冲突
    var_key = f"task_timeout_{context['dag_run'].run_id}"
    Variable.set(var_key, timeout)
    # 通过XCom传递Variable键给后续任务
    context["ti"].xcom_push(key="timeout_var_key", value=var_key)

def main_process(**context):
    # 从XCom获取Variable键并读取超时时间
    var_key = context["ti"].xcom_pull(key="timeout_var_key")
    timeout_minutes = int(Variable.get(var_key))
    
    # 编写核心任务逻辑，可结合内部超时判断
    print(f"任务超时设置为 {timeout_minutes} 分钟")
    
    # 任务完成后清理临时Variable
    Variable.delete(var_key)

with DAG(
    dag_id="dynamic_timeout_via_variable",
    start_date=datetime(2024, 1, 1),
    params={"mls_sid": "0", "time_out": 10},
    schedule=None
) as dag:
    set_timeout = PythonOperator(
        task_id="set_timeout_variable",
        python_callable=set_timeout_var,
        provide_context=True
    )
    
    main_task = PythonOperator(
        task_id="main_process_task",
        python_callable=main_process,
        provide_context=True,
        # 设置兜底的最大超时时间
        execution_timeout=timedelta(hours=1)
    )
    
    set_timeout >> main_task

方案3：动态生成任务实例（局限性较大） #

利用TaskFlow API动态生成任务，在生成时根据params设置execution_timeout。但该方案依赖DAG解析阶段的模板渲染，触发时修改的参数需要等待DAG重新解析才能生效，仅适合对实时性要求不高的场景。

示例代码：

from airflow.decorators import dag, task
from datetime import datetime, timedelta

@dag(
    start_date=datetime(2024, 1, 1),
    params={"mls_sid": "0", "time_out": 10},
    schedule=None
)
def dynamic_task_timeout_dag():
    def create_task_with_timeout(timeout_minutes):
        @task(execution_timeout=timedelta(minutes=timeout_minutes))
        def run_task(params):
            print(f"处理数据子集 {params['mls_sid']}")
            # 核心任务逻辑
        return run_task
    
    # 从默认params中读取超时时间生成任务
    default_timeout = int("{{ params.time_out }}")
    task_instance = create_task_with_timeout(default_timeout)
    task_instance()

dag = dynamic_task_timeout_dag()

内容的提问来源于stack exchange，提问作者Chris Lichliter