我使用的是 Airflow 1.10.2，但 Airflow 似乎忽略了我为 DAG 设置的超时。
我正在使用以下命令为 DAG 设置超时期限dagrun_timeout参数（例如 20 秒），我有一个需要 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
不会抛出任何错误消息。我使用了不正确的参数吗？
如中所述源代码 https://github.com/apache/airflow/blob/master/airflow/models/dag.py#L138-L141:
: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 https://github.com/apache/airflow/blob/master/airflow/models/baseoperator.py#L167-L168所有运营商均可用。
例如，您可以在您的设备上设置 60 秒超时PythonOperator像这样：
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)
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)
