ट्रिगर्स और फॉल्ट टॉलरेंस

Python में Apache Airflow परिचय

Mike Metzger

Data Engineer

मजबूत पाइपलाइंस बनाना

ट्रिगर्स - नियंत्रित करते हैं कि टास्क कब चलें
- विशेष शर्तों पर चलने वाले टास्क परिभाषित करें
- डिफॉल्ट ट्रिगर - all success

ट्रिगर्स जो तय करते हैं कि टास्क कब चलें, उसका चित्रण

फॉल्ट टॉलरेंस - टास्क के भीतर रिकवरी सक्षम करता है
- व्यक्तिगत टास्क को रीट्राई करें

फॉल्ट टॉलरेंस जो असफल टास्क को रीट्राई करता है, उसका चित्रण

ट्रिगर रूल्स

टास्क शुरू होने से पहले पिछले टास्क की state जाँचें
यह तय करने दें कि टास्क आगे बढ़ें या नहीं
डिफॉल्ट: आगे बढ़ने से पहले सभी पिछले टास्क सफल होने चाहिए
ट्रिगर रूल्स से आप इसे बदल सकते हैं

अपस्ट्रीम टास्क की state जाँचते ट्रिगर रूल्स का चित्रण

ट्रिगर रूल्स के उपयोग

नोटिफिकेशन टास्क
क्लीनअप टास्क
कंडीशनल execution

नोटिफिकेशन और क्लीनअप जैसे ट्रिगर रूल उपयोग मामलों का चित्रण

मुख्य ट्रिगर रूल्स

all_success - सभी पिछले टास्क सफल, कोई skip नहीं
all_failed - सभी पिछले टास्क असफल
all_done - सभी पिछले टास्क पूरे, परिणाम चाहे कुछ भी हो
one_failed - कम से कम एक टास्क असफल
one_success - कम से कम एक टास्क सफल
none_failed - सभी पिछले टास्क सफल या skipped

ट्रिगर रूल implementation

from airflow.utils.trigger_rule import TriggerRule
ट्रिगर रूल्स @task डेकोरेटर पर लगाए जाते हैं
trigger_rule=TriggerRule.<TriggerRule enum>

@task(trigger_rule=TriggerRule.ALL_SUCCESS)
def run_if_everything_succeeds:
  print('All previous tasks succeeded!')

फॉल्ट टॉलरेंस attributes

retries - टास्क पैरामीटर जो बताता है कि Airflow असफल टास्क को फेल मार्क करने से पहले कितनी बार रीट्राई करे
retry_delay - टास्क पैरामीटर, timedelta लेता है जो रीट्राई के बीच wait time तय करता है

@task(
  trigger_rule=TriggerRule.ONE_SUCCESS,
  retries=3,
  retry_delay=timedelta(minutes=5)
)

TriggerDagRunOperator

एक Dag दूसरे Dag को शुरू कर सकता है
कोड और Dags उपलब्ध कराता है
from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator
trigger_dag_id - dag_id से मिलना चाहिए
wait_for_completion - पूरा होने तक आगे के टास्क रोकें
poke_interval - कितनी बार जाँचें
conf - child Dag को पास होने वाला डेटा

TriggerDagRun उदाहरण

trigger_child = TriggerDagRunOperator(
        task_id="trigger_child_pipeline",
        trigger_dag_id="child_pipeline_dag",   
        wait_for_completion=True,              
        poke_interval=30,                      
        conf={                                 
            "source": "s3://my-bucket/raw/",
        },
    )

task1 >> trigger_child    # पहले टास्क चलाएँ, फिर child Dag शुरू करें
trigger_child >> cleanup  # child Dag पूरे होने पर cleanup टास्क चलाएँ

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

Python में Apache Airflow परिचय