时间戳

首页 > 解决方案 > 如何为 Apache Airflow DAG 定义超时？

airflow - 如何为 Apache Airflow DAG 定义超时？

问题描述

我正在使用 Airflow 1.10.2，但 Airflow 似乎忽略了我为 DAG 设置的超时。

我正在使用dagrun_timeout参数（例如 20 秒）为 DAG 设置超时时间，并且我有一个需要 2 分钟才能运行的任务，但 Airflow 将 DAG 标记为成功！

args = {
    'owner': 'me',
    'start_date': airflow.utils.dates.days_ago(2),
    'provide_context': True,
}

dag = DAG(
    'test_timeout',
     schedule_interval=None,
     default_args=args,
     dagrun_timeout=timedelta(seconds=20),
)

def this_passes(**kwargs):
    return

def this_passes_with_delay(**kwargs):
    time.sleep(120)
    return

would_succeed = PythonOperator(
    task_id='would_succeed',
    dag=dag,
    python_callable=this_passes,
    email=to,
)

would_succeed_with_delay = PythonOperator(
    task_id='would_succeed_with_delay',
    dag=dag,
    python_callable=this_passes_with_delay,
    email=to,
)

would_succeed >> would_succeed_with_delay

不会抛出任何错误消息。我是否使用了不正确的参数？

标签： airflow

解决方案

如源代码中所述：

:param dagrun_timeout: specify how long a DagRun should be up before
    timing out / failing, so that new DagRuns can be created. The timeout
    is only enforced for scheduled DagRuns, and only once the
    # of active DagRuns == max_active_runs.

所以这可能是您设置时的预期行为schedule_interval=None。这里的想法是确保计划的 DAG 不会永远持续并阻止后续运行实例。

现在，您可能对execution_timeout可用的所有运算符感兴趣。例如，您可以PythonOperator像这样设置 60 秒超时：

would_succeed_with_delay = PythonOperator(task_id='would_succeed_with_delay',
                            dag=dag,
                            execution_timeout=timedelta(seconds=60),
                            python_callable=this_passes_with_delay,
                            email=to)

推荐阅读