ڈی اے جی رن کنفیگریشن کا استعمال کرتے ہوئے ایئر فلو میں متحرک ٹاسک تسلسل پیدا کرنا

ہوا کے بہاؤ میں متحرک ٹاسک انحصار کی طاقت کو غیر مقفل کرنا

اپاچی ایئر فلو ایک طاقتور ورک فلو آٹومیشن ٹول ہے ، لیکن متحرک انحصار کو سنبھالنا بعض اوقات کسی پہیلی کو حل کرنے کی طرح محسوس کرسکتا ہے۔ جب ہدایت شدہ ایسائکلک گراف (ڈی اے جی) کو ڈیزائن کرتے ہو تو ، ہارڈ کوڈنگ ٹاسک کی ترتیب آسان استعمال کے معاملات کے ل work کام کر سکتی ہے ، لیکن اگر اس ڈھانچے کو رن ٹائم کے وقت طے کرنے کی ضرورت ہو تو کیا ہوگا؟ 🤔

ذرا تصور کریں کہ آپ ڈیٹا پائپ لائن پر کام کر رہے ہیں جہاں کاموں کو پھانسی دینے کے لئے آنے والے ڈیٹا پر منحصر ہے۔ مثال کے طور پر ، روزانہ کی ترتیب پر مبنی فائلوں کے مختلف سیٹوں پر کارروائی کرنا یا کاروباری اصول پر مبنی متغیر تبدیلیوں کو عملی جامہ پہنانا۔ ایسے معاملات میں ، ایک مستحکم ڈی اے جی اسے کاٹ نہیں پائے گی - آپ کو انحصار کی وضاحت کے لئے ایک طریقہ کی ضرورت ہے۔

یہ بالکل وہی ہے جہاں ایئر فلو کا ہے dag_run.conf گیم چینجر ہوسکتا ہے۔ ڈی اے جی کو متحرک کرتے وقت کنفیگریشن لغت کو پاس کرکے ، آپ متحرک طور پر ٹاسک سلسلوں کو تیار کرسکتے ہیں۔ تاہم ، اس کو منظم طریقے سے نافذ کرنے کے لئے ایئر فلو کے عمل درآمد کے ماڈل کی گہری تفہیم کی ضرورت ہے۔

اس مضمون میں ، ہم یہ دریافت کریں گے کہ متحرک ڈی اے جی کی تعمیر کیسے کی جائے جہاں رن ٹائم کے استعمال میں ٹاسک انحصار کا تعین کیا جائے dag_run.conf. اگر آپ اس کو حاصل کرنے کے لئے جدوجہد کر رہے ہیں اور کوئی واضح حل نہیں ملا ہے تو ، فکر نہ کریں - آپ تنہا نہیں ہیں! آئیے عملی مثالوں کے ساتھ اسے قدم بہ قدم توڑ دیں۔ 🚀

ہوا کے بہاؤ میں رن ٹائم کنفیگریشن کے ساتھ متحرک ڈی اے جی کی تعمیر

اپاچی ایئر فلو پیچیدہ ورک فلوز کو آرکیسٹریٹ کرنے کے لئے ایک طاقتور ٹول ہے ، لیکن اس کی اصل طاقت اس کی لچک میں ہے۔ اس سے پہلے پیش کردہ اسکرپٹ یہ ظاہر کرتے ہیں کہ کیسے بنائیں متحرک ڈی اے جی جہاں کام کے انحصار کا استعمال کرتے ہوئے رن ٹائم کے وقت طے کیا جاتا ہے dag_run.conf. عمل کے ل عناصر کی فہرست کو سخت کوڈ کرنے کے بجائے ، ڈی اے جی متحرک طور پر ان کو بازیافت کرتا ہے جب متحرک ہوجاتا ہے ، اور مزید موافقت پذیر ورک فلوز کی اجازت دیتا ہے۔ یہ خاص طور پر حقیقی دنیا کے منظرناموں میں مفید ہے ، جیسے متغیر ڈیٹاسیٹس پر کارروائی کرنا یا بیرونی حالات کی بنیاد پر مخصوص کاموں کو انجام دینا۔ ایک ETL پائپ لائن کا تصور کریں جہاں فائلوں کو روزانہ تبدیل کرنے پر کارروائی کرنے کے لئے - یہ نقطہ نظر آٹومیشن کو بہت آسان بنا دیتا ہے۔ 🚀

پہلا اسکرپٹ استعمال کرتا ہے پائیٹونپریٹر کاموں کو عملی جامہ پہنانے اور انحصار کو متحرک طور پر طے کرنے کے لئے۔ یہ عناصر کی فہرست سے نکالتا ہے dag_run.conf، اس بات کو یقینی بنانا کہ جب ضرورت ہو تب ہی کام بنائے جائیں۔ فہرست میں ہر عنصر ایک انوکھا کام بن جاتا ہے ، اور انحصار ترتیب سے ترتیب دیا جاتا ہے۔ دوسرا نقطہ نظر فائدہ اٹھاتا ہے ٹاسک فلو API، جو ڈیگ تخلیق کو سجاوٹ کے ساتھ آسان بناتا ہے @ڈیگ اور @ٹاسک. یہ طریقہ ڈی اے جی کو زیادہ پڑھنے کے قابل بناتا ہے اور کلینر پر عمل درآمد کی منطق کو برقرار رکھتا ہے۔ یہ نقطہ نظر اس بات کو یقینی بناتے ہیں کہ ورک فلوز کوڈ کی تبدیلیوں کی ضرورت کے بغیر مختلف ترتیبوں کے مطابق ڈھال سکتے ہیں۔

مثال کے طور پر ، ایک ایسے منظر نامے پر غور کریں جہاں ایک ای کامرس کمپنی بیچوں میں آرڈر پر کارروائی کرتی ہے۔ کچھ دن دوسروں کے مقابلے میں زیادہ فوری احکامات ہوسکتے ہیں ، جس میں مختلف ٹاسک سلسلے کی ضرورت ہوتی ہے۔ جامد ڈی اے جی کے استعمال کا مطلب یہ ہوگا کہ ہر بار ترجیحات میں تبدیلی کی جائے۔ ہمارے متحرک ڈی اے جی نقطہ نظر کے ساتھ ، ایک بیرونی نظام ڈی اے جی کو ایک مخصوص ٹاسک تسلسل کے ساتھ متحرک کرسکتا ہے ، جس سے عمل کو زیادہ موثر بنایا جاسکتا ہے۔ ایک اور استعمال کا معاملہ ڈیٹا سائنس میں ہے ، جہاں ماڈلز کو آنے والے ڈیٹا کی تقسیم کی بنیاد پر دوبارہ تربیت کی ضرورت ہوسکتی ہے۔ مطلوبہ ماڈل کی ترتیب کو متحرک طور پر پاس کرکے ، وقت اور وسائل کی بچت کرنے والے صرف ضروری کمپیوٹوں کو ہی انجام دیا جاتا ہے۔ 🎯

خلاصہ یہ کہ یہ اسکرپٹ رن ٹائم ان پٹ پر مبنی ڈی اے جی کو متحرک طور پر پیدا کرنے کے لئے ایک بنیاد فراہم کرتے ہیں۔ فائدہ اٹھا کر ایئر فلو کا ٹاسک فلو API یا روایتی پائیٹونوپریٹر نقطہ نظر ، ڈویلپر لچکدار ، ماڈیولر اور موثر ورک فلو تشکیل دے سکتے ہیں۔ اس سے دستی مداخلت کی ضرورت ختم ہوجاتی ہے اور دوسرے آٹومیشن سسٹم کے ساتھ ہموار انضمام کی اجازت ملتی ہے۔ چاہے کسٹمر کے احکامات پر کارروائی کریں ، ڈیٹا پائپ لائنوں کا انتظام ، یا کلاؤڈ ورک فلوز کو آرکیسٹریٹنگ ، متحرک ڈی اے جی مخصوص کاروباری ضروریات کے مطابق ہوشیار آٹومیشن کو قابل بنائے۔

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، جو پہلے سے طے شدہ منطق کی بنیاد پر عملدرآمد کرنے کے لئے اگلے کام کا تعین کرتا ہے۔ فرض کریں کہ ہمارے پاس ایک متحرک ڈی اے جی ہے جو فائلوں پر کارروائی کرتی ہے ، لیکن صرف ایک مخصوص سائز سے اوپر کی فائلوں کو توثیق کی ضرورت ہوتی ہے۔ ترتیب سے تمام کاموں کو عملی جامہ پہنانے کے بجائے ، ہم متحرک طور پر فیصلہ کرسکتے ہیں کہ کون سے کام چلائیں ، عملدرآمد کا وقت بہتر بنائیں اور وسائل کے استعمال کو کم کریں۔ یہ نقطہ نظر اس بات کو یقینی بناتا ہے کہ صرف متعلقہ ورک فلوز کو متحرک کیا جاتا ہے ، جس سے ڈیٹا پائپ لائنوں کو زیادہ موثر بنایا جاتا ہے۔ 🚀

متحرک ڈی اے جی کو بڑھانے کا دوسرا طریقہ شامل کرنا ہے XComs (کراس کمیونیکیشن پیغامات)۔ XCOMS کاموں کو ڈیٹا کا تبادلہ کرنے کی اجازت دیتا ہے ، اس کا مطلب یہ ہے کہ متحرک طور پر تیار کردہ ٹاسک تسلسل اقدامات کے مابین معلومات کو منتقل کرسکتا ہے۔ مثال کے طور پر ، ای ٹی ایل پائپ لائن میں ، ایک پری پروسیسنگ ٹاسک مطلوبہ تبدیلیوں کا تعین کرسکتا ہے اور ان تفصیلات کو بعد کے کاموں تک پہنچا سکتا ہے۔ یہ طریقہ واقعی میں ڈیٹا سے چلنے والے ورک فلوز کو قابل بناتا ہے ، جہاں پر عملدرآمد کا بہاؤ ریئل ٹائم ان پٹ پر مبنی موافقت پذیر ہوتا ہے ، جس سے آٹومیشن کی صلاحیتوں میں نمایاں اضافہ ہوتا ہے۔