[Airflow] retry와 상관없이 특정 시간 이후 task fail시키기

개요

에어플로우에서는 task가 실패했을 때 재실행 하는 방법을 제공하고 있다.

재실행은 다음과 같이 설정할 수 있다.

 

import pendulum

from airflow import DAG
from airflow.operators.bash import BashOperator

from datetime import timedelta

default_args = {
    'owner': 'heenj',
    'retries': 6, # 재실행 횟수
    'retry_delay': timedelta(minutes=10) # 재실행 대기 시간
}

with DAG(
    dag_id='retry_dag',
    default_args=default_args,
    start_date=datetime(2024, 3, 1, tzinfo=pendulum.timezone("Asia/Seoul")),
    schedule_interval= "0 21 * * *",
    tags=['example'],
    catchup=False
) as dag:

	aggregate_db_message_job = BashOperator(
        task_id='test_job',
        bash_command= "echo 3")

• 재실행 횟수가 6, 대기 시간을 10분으로 설정

 

문제 원인

여기서 재실행 횟수와 시간은 task가 실행되는 시간을 제외하고, 실패 시 온전히 대기하는 시간이다.

따라서 해당 task가 모두 실패하면 실제 대기 시간은 60분을 초과할 수 있다.

1시간 후까지 해당 dag이 종료되지 않으면 에러가 발생하는 방법이 필요했다.

 

 

해결 방법

최대 실행시간 (execution_timeout)을 따로 설정해줄 수 있다.

default_args에 다음 파라미터를 추가해주자.

default_args = {
    'owner': 'heenj',
    'retries': 6, # 재실행 횟수
    'retry_delay': timedelta(minutes=10) # 재실행 대기 시간
    'execution_timeout': timedelta(hours=1)  # DAG의 최대 실행 시간, 넘어갈 시 fail 처리
}