फेल्योर हैंडलिंग और रिट्राइ के साथ मज़बूत Dags

Airflow के साथ Data Pipelines बनाना

Volker Janz

Senior Developer Advocate at Astronomer

टास्क क्यों फेल होते हैं

टास्क कनेक्शन समस्याएँ

API timeouts और rate limits
छोटे network interruptions और domain name resolution delays
एक ही resource के लिए कई tasks की competition

रिट्राइज़

@task(
    retries=3,
    retry_delay=timedelta(minutes=2),
)
def fetch_weather():
    response = requests.get("https://api.weather.com/forecast")
    response.raise_for_status()
    return response.json()

फेल होने पर टास्क अधिकतम 3 बार retry करेगा
हर प्रयास के बीच 2 मिनट रुकेगा

Exponential backoff

@task(
    retries=3,
    retry_delay=timedelta(minutes=2),
    retry_exponential_backoff=2.0,
)
def fetch_weather():
    ...

Exponential backoff

हर retry के बाद delay दोगुना होता है
बाहरी सेवाओं को recover करने का समय मिलता है

on_failure_callback

def alert_on_failure(context):
    dag_id = context["dag"].dag_id
    task_id = context["ti"].task_id
    print(f"ALERT: {task_id} in {dag_id} failed!")

@task(on_failure_callback=alert_on_failure)
def fetch_weather():
    ...

context dict: dag, ti, exception, log URL
Slack, PagerDuty, या किसी भी alerting tool को route करें

Dag बनाम task स्तर पर callbacks

Dag-स्तर (catch-all)

@dag(
    on_failure_callback=alert_team,
)
def my_pipeline():
    ...

Task-स्तर (specific)

@task(
    on_failure_callback=page_oncall,
)
def critical_step():
    ...

Task-स्तर Dag-स्तर को override करता है
लेयर्ड alerting के लिए दोनों इस्तेमाल करें

max_consecutive_failed_dag_runs

@dag(max_consecutive_failed_dag_runs=3)
def monitoring_pipeline():
    ...

Failed Dag runs followed by auto-pause

लगातार फेल्योर से होने वाली alert fatigue रोकता है
मैन्युअली unpause करने पर Dag फिर से चलता है

सबको साथ में जोड़ना

@dag(
    max_consecutive_failed_dag_runs=3,
    on_failure_callback=alert_team,
)
def weather_pipeline():

    @task(
        retries=3,
        retry_delay=timedelta(minutes=2),
        retry_exponential_backoff=True,
        on_failure_callback=page_oncall,
    )
    def fetch_weather():
        ...

Retries + backoff अस्थायी फेल्योर संभालते हैं
Callbacks सही लोगों को alert करते हैं
Auto-pause अनावश्यक शोर रोकता है

अभ्यास करते हैं!

Airflow के साथ Data Pipelines बनाना