DAG & Backfill

from airflow import DAG from airflow.operators.python_operator import PythonOperator from datetime import datetime, timedelta from error_alarm import run_slack_alarm default_args = { 'owner': 'airflow', 'depends_on_past': False, 'start_date': datetime(2024, 1, 1), 'email_on_failure': False, 'email_on_retry': False, 'retries': 3, # 재시도 횟수 'retry_delay': timedelta(minutes=5), # 재시도 간격 } # 실패 시 호출될 함수 def on_failure_callback(context): run_slack_alarm(context) return # DAG 정의 dag = DAG( 'retry_and_recovery_example', default_args=default_args, description='A simple DAG with retry and recovery', schedule_interval=timedelta(days=1), ) # 재시도 및 복구가 필요한 작업 정의 def my_task_function(): # 예외 발생 시 자동으로 재시도됩니다. raise Exception("An error occurred") task_with_retry_and_recovery = PythonOperator( task_id='task_with_retry_and_recovery', python_callable=my_task_function, retries=3, # 작업 별 재시도 횟수 retry_delay=timedelta(minutes=5), # 작업 별 재시도 간격 on_failure_callback=on_failure_callback, # 실패 시 호출될 함수 dag=dag, )