DAG रन कॉन्फ़िगरेशन का उपयोग करके एयरफ्लो में डायनेमिक टास्क सीक्वेंस उत्पन्न करना

एयरफ्लो में गतिशील कार्य निर्भरता की शक्ति को अनलॉक करना

Apache Airflow एक शक्तिशाली वर्कफ़्लो ऑटोमेशन टूल है, लेकिन गतिशील निर्भरता को संभालने से कभी -कभी एक पहेली को हल करने जैसा महसूस हो सकता है। एक निर्देशित एसाइक्लिक ग्राफ (डीएजी) डिजाइन करते समय, हार्डकोडिंग टास्क सीक्वेंस सरल उपयोग के मामलों के लिए काम कर सकते हैं, लेकिन क्या होगा यदि संरचना को रनटाइम पर निर्धारित करने की आवश्यकता है? 🤔

कल्पना कीजिए कि आप एक डेटा पाइपलाइन पर काम कर रहे हैं जहां निष्पादित किए जाने वाले कार्य आने वाले डेटा पर निर्भर करते हैं। उदाहरण के लिए, एक दैनिक कॉन्फ़िगरेशन के आधार पर फ़ाइलों के विभिन्न सेटों को संसाधित करना या व्यावसायिक नियम के आधार पर चर परिवर्तनों को निष्पादित करना। ऐसे मामलों में, एक स्थैतिक डीएजी ने इसे काट नहीं दिया - आपको गतिशील रूप से निर्भरता को परिभाषित करने का एक तरीका चाहिए।

यह ठीक है जहां एयरफ्लो है dag_run.conf एक गेम-चेंजर हो सकता है। DAG को ट्रिगर करते समय एक कॉन्फ़िगरेशन शब्दकोश पास करके, आप गतिशील रूप से कार्य अनुक्रम उत्पन्न कर सकते हैं। हालांकि, इसे एक संरचित तरीके से लागू करने के लिए एयरफ्लो के निष्पादन मॉडल की गहरी समझ की आवश्यकता होती है।

इस लेख में, हम यह पता लगाएंगे कि एक गतिशील DAG का निर्माण कैसे करें जहां कार्य निर्भरताएं रनटाइम का उपयोग करके निर्धारित की जाती हैं dag_run.conf। यदि आप इसे प्राप्त करने के लिए संघर्ष कर रहे हैं और स्पष्ट समाधान नहीं मिला है, तो चिंता न करें - आप अकेले नहीं हैं! चलो व्यावहारिक उदाहरणों के साथ इसे नीचे कदम से तोड़ दें। 🚀

एयरफ्लो में रनटाइम कॉन्फ़िगरेशन के साथ डायनामिक डैग का निर्माण

अपाचे एयरफ्लो कॉम्प्लेक्स वर्कफ़्लोज़ को ऑर्केस्ट्रेट करने के लिए एक शक्तिशाली उपकरण है, लेकिन इसकी वास्तविक ताकत इसके लचीलेपन में निहित है। पहले प्रस्तुत की गई स्क्रिप्ट यह प्रदर्शित करती है कि कैसे बनाएं गतिशील दाग जहां कार्य निर्भरताएं रनटाइम का उपयोग करके निर्धारित की जाती हैं dag_run.conf। संसाधित करने के लिए तत्वों की सूची को हार्डकोड करने के बजाय, डीएजी ट्रिगर होने पर उन्हें गतिशील रूप से पुनर्प्राप्त करता है, अधिक अनुकूलनीय वर्कफ़्लो के लिए अनुमति देता है। यह वास्तविक दुनिया के परिदृश्यों में विशेष रूप से उपयोगी है, जैसे कि चर डेटासेट को संसाधित करना या बाहरी स्थितियों के आधार पर विशिष्ट कार्यों को निष्पादित करना। एक ईटीएल पाइपलाइन की कल्पना करें जहां दैनिक रूप से बदलने के लिए फाइलें बदलती हैं - यह दृष्टिकोण स्वचालन को बहुत आसान बनाता है। 🚀

पहली स्क्रिप्ट का उपयोग करता है पायथनपरक कार्यों को निष्पादित करने और गतिशील रूप से निर्भरता निर्धारित करने के लिए। यह तत्वों की सूची को निकालता है dag_run.conf, यह सुनिश्चित करना कि कार्यों को केवल जरूरत पड़ने पर बनाया जाता है। सूची में प्रत्येक तत्व एक अद्वितीय कार्य बन जाता है, और निर्भरता क्रमिक रूप से सेट की जाती है। दूसरा दृष्टिकोण का लाभ उठाता है टास्कफ्लो एपीआई, जो कि डेग सृजन को सजाने वाले के साथ सरल करता है @dag और @काम। यह विधि डीएजी को अधिक पठनीय बनाती है और क्लीनर निष्पादन तर्क को बनाए रखती है। ये दृष्टिकोण यह सुनिश्चित करते हैं कि वर्कफ़्लोज़ कोड परिवर्तनों की आवश्यकता के बिना विभिन्न कॉन्फ़िगरेशन के लिए अनुकूल हो सकते हैं।

उदाहरण के लिए, एक परिदृश्य पर विचार करें जहां एक ई-कॉमर्स कंपनी बैचों में आदेशों को संसाधित करती है। कुछ दिनों में दूसरों की तुलना में अधिक जरूरी आदेश हो सकते हैं, विभिन्न कार्य अनुक्रमों की आवश्यकता होती है। एक स्थिर DAG का उपयोग करने का मतलब होगा कि हर बार प्राथमिकताएं बदलें कोड को संशोधित करना। हमारे गतिशील DAG दृष्टिकोण के साथ, एक बाहरी प्रणाली DAG को एक विशिष्ट कार्य अनुक्रम के साथ ट्रिगर कर सकती है, जिससे प्रक्रिया अधिक कुशल हो जाती है। एक अन्य उपयोग मामला डेटा विज्ञान में है, जहां मॉडल को आने वाले डेटा वितरण के आधार पर रिट्रेनिंग की आवश्यकता हो सकती है। आवश्यक मॉडल कॉन्फ़िगरेशन को गतिशील रूप से पारित करके, केवल आवश्यक संगणना को निष्पादित किया जाता है, समय और संसाधनों की बचत होती है। 🎯

सारांश में, ये स्क्रिप्ट रनटाइम इनपुट के आधार पर गतिशील रूप से उत्पन्न करने वाले डीएजी के लिए एक नींव प्रदान करते हैं। लाभ उठाकर AirFlow का टास्कफ्लो एपीआई या पारंपरिक पायथनऑपरेटर दृष्टिकोण, डेवलपर्स लचीले, मॉड्यूलर और कुशल वर्कफ़्लोज़ बना सकते हैं। यह मैनुअल हस्तक्षेप की आवश्यकता को समाप्त करता है और अन्य स्वचालन प्रणालियों के साथ सहज एकीकरण के लिए अनुमति देता है। चाहे ग्राहक ऑर्डर, डेटा पाइपलाइनों का प्रबंधन, या क्लाउड वर्कफ़्लोज़ को ऑर्केस्ट्रेट करना, डायनेमिक DAGs विशिष्ट व्यावसायिक आवश्यकताओं के अनुरूप होशियार स्वचालन को सक्षम करते हैं।

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago
from airflow.models import DagRun
import json
# Define default args
default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': days_ago(1),
}
# Function to process each element
def process_element(element, kwargs):
    print(f"Processing element: {element}")
# Define DAG
dag = DAG(
    'dynamic_task_dag',
    default_args=default_args,
    schedule_interval=None,
)
# Extract elements from dag_run.conf
def generate_tasks(kwargs):
    conf = kwargs.get('dag_run').conf or {}
    elements = conf.get('elements', [])
    task_list = []
    for i, group in enumerate(elements):
        for j, element in enumerate(group):
            task_id = f"process_element_{i}_{j}"
            task = PythonOperator(
                task_id=task_id,
                python_callable=process_element,
                op_args=[element],
                dag=dag,
            )
            task_list.append(task)
    return task_list
# Generate dynamic tasks
tasks = generate_tasks()
# Define dependencies dynamically
for i in range(len(tasks) - 1):
    tasks[i + 1].set_upstream(tasks[i])

from airflow.decorators import dag, task
from datetime import datetime
# Define DAG
@dag(schedule_interval=None, start_date=datetime(2025, 1, 28), catchup=False)
def dynamic_taskflow_dag():
    @task
    def process_element(element: str):
        print(f"Processing {element}")
    @task
    def extract_elements(dag_run=None):
        conf = dag_run.conf or {}
        return conf.get('elements', [])
    elements = extract_elements()
    task_groups = [[process_element(element) for element in group] for group in elements]
    # Define dependencies dynamically
    for i in range(len(task_groups) - 1):
        for upstream_task in task_groups[i]:
            for downstream_task in task_groups[i + 1]:
                downstream_task.set_upstream(upstream_task)
dynamic_taskflow_dag()

एयरफ्लो में सशर्त निष्पादन के साथ गतिशील कार्य अनुक्रमण को बढ़ाना

एक शक्तिशाली अभी तक अक्सर अनदेखी की गई सुविधा में अपाचे एयरफ्लो सशर्त निष्पादन है, जो गतिशील कार्य अनुक्रमण के लचीलेपन में सुधार कर सकता है। से कार्य निर्भरता को पुनः प्राप्त करते हुए dag_run.conf उपयोगी है, वास्तविक दुनिया के परिदृश्यों को अक्सर विशिष्ट स्थितियों के आधार पर केवल कुछ कार्यों को निष्पादित करने की आवश्यकता होती है। उदाहरण के लिए, कुछ डेटासेट को विश्लेषण से पहले प्रीप्रोसेसिंग की आवश्यकता हो सकती है, जबकि अन्य को सीधे संसाधित किया जा सकता है।

एयरफ्लो में सशर्त निष्पादन का उपयोग करके लागू किया जा सकता है BranchPythonOperator, जो पूर्वनिर्धारित तर्क के आधार पर निष्पादित करने के लिए अगला कार्य निर्धारित करता है। मान लीजिए कि हमारे पास एक गतिशील DAG है जो फ़ाइलों को संसाधित करता है, लेकिन केवल एक निश्चित आकार से ऊपर की फ़ाइलों को सत्यापन की आवश्यकता होती है। क्रमिक रूप से सभी कार्यों को निष्पादित करने के बजाय, हम गतिशील रूप से तय कर सकते हैं कि कौन से कार्यों को चलाना है, निष्पादन समय का अनुकूलन करना और संसाधन उपयोग को कम करना है। यह दृष्टिकोण यह सुनिश्चित करता है कि केवल प्रासंगिक वर्कफ़्लो को ट्रिगर किया जाता है, जिससे डेटा पाइपलाइन अधिक कुशल हो जाती है। 🚀

गतिशील dags को बढ़ाने का एक और तरीका शामिल है XComs (क्रॉस-कम्युनिकेशन मैसेज)। XCOM कार्यों को डेटा का आदान -प्रदान करने की अनुमति देते हैं, जिसका अर्थ है कि एक गतिशील रूप से बनाया गया कार्य अनुक्रम चरणों के बीच जानकारी पारित कर सकता है। उदाहरण के लिए, एक ईटीएल पाइपलाइन में, एक प्रीप्रोसेसिंग कार्य आवश्यक परिवर्तनों को निर्धारित कर सकता है और उन विवरणों को बाद के कार्यों के लिए पारित कर सकता है। यह विधि वास्तव में डेटा-संचालित वर्कफ़्लोज़ को सक्षम करती है, जहां निष्पादन प्रवाह वास्तविक समय के इनपुट के आधार पर अनुकूलता करता है, स्वचालन क्षमताओं को काफी बढ़ाता है।