ডিএজি রান কনফিগারেশন ব্যবহার করে এয়ারফ্লোতে গতিশীল টাস্ক সিকোয়েন্সগুলি তৈরি করা

বায়ুপ্রবাহে গতিশীল টাস্ক নির্ভরতাগুলির শক্তি আনলক করা

অ্যাপাচি এয়ারফ্লো একটি শক্তিশালী ওয়ার্কফ্লো অটোমেশন সরঞ্জাম, তবে গতিশীল নির্ভরতাগুলি পরিচালনা করা কখনও কখনও ধাঁধা সমাধানের মতো অনুভব করতে পারে। একটি নির্দেশিত অ্যাসাইক্লিক গ্রাফ (ডিএজি) ডিজাইন করার সময়, হার্ডকোডিং টাস্ক সিকোয়েন্সগুলি সাধারণ ব্যবহারের ক্ষেত্রে কাজ করতে পারে, তবে যদি রানটাইমে কাঠামোটি নির্ধারণ করা দরকার? 🤔

কল্পনা করুন যে আপনি কোনও ডেটা পাইপলাইনে কাজ করছেন যেখানে কার্যকর করা কাজগুলি আগত ডেটার উপর নির্ভর করে। উদাহরণস্বরূপ, একটি দৈনিক কনফিগারেশনের উপর ভিত্তি করে ফাইলগুলির বিভিন্ন সেট প্রক্রিয়াজাতকরণ বা ব্যবসায়ের নিয়মের ভিত্তিতে ভেরিয়েবল ট্রান্সফর্মেশনগুলি সম্পাদন করা। এই জাতীয় ক্ষেত্রে, একটি স্ট্যাটিক ডিএজি এটি কাটবে না - আপনার গতিশীলভাবে নির্ভরতা সংজ্ঞায়িত করার জন্য একটি উপায় প্রয়োজন।

এটি ঠিক যেখানে এয়ারফ্লো এর dag_run.conf গেম-চেঞ্জার হতে পারে। কোনও ডিএজি ট্রিগার করার সময় একটি কনফিগারেশন ডিকশনারি পাস করে আপনি গতিশীলভাবে টাস্ক সিকোয়েন্সগুলি তৈরি করতে পারেন। যাইহোক, এটি একটি কাঠামোগত উপায়ে বাস্তবায়নের জন্য এয়ারফ্লোর এক্সিকিউশন মডেলটির গভীর বোঝার প্রয়োজন।

এই নিবন্ধে, আমরা কীভাবে একটি গতিশীল ডিএজি তৈরি করতে পারি যেখানে টাস্ক নির্ভরতাগুলি রানটাইম ব্যবহার করে নির্ধারিত হয় তা অনুসন্ধান করব dag_run.conf। আপনি যদি এটি অর্জনের জন্য লড়াই করে যাচ্ছেন এবং একটি পরিষ্কার সমাধান খুঁজে না পেয়ে থাকেন তবে চিন্তা করবেন না - আপনি একা নন! ব্যবহারিক উদাহরণ সহ ধাপে ধাপে এটিকে ভেঙে দিন। 🚀

এয়ারফ্লোতে রানটাইম কনফিগারেশন সহ গতিশীল ডিএজিএস বিল্ডিং

অ্যাপাচি এয়ারফ্লো জটিল কর্মপ্রবাহকে অর্কেস্টেটিংয়ের জন্য একটি শক্তিশালী সরঞ্জাম, তবে এর সত্য শক্তি তার নমনীয়তার মধ্যে রয়েছে। পূর্বে উপস্থাপিত স্ক্রিপ্টগুলি কীভাবে তৈরি করবেন তা প্রমাণ করে ডায়নামিক ড্যাগ যেখানে টাস্ক নির্ভরতাগুলি রানটাইম ব্যবহার করে নির্ধারিত হয় dag_run.conf। প্রক্রিয়া করার জন্য উপাদানগুলির তালিকাটি হার্ডকোডিংয়ের পরিবর্তে, ডিএজি ট্রিগার করার সময় তাদের গতিশীলভাবে পুনরুদ্ধার করে, আরও অভিযোজিত কর্মপ্রবাহের জন্য অনুমতি দেয়। এটি রিয়েল-ওয়ার্ল্ড দৃশ্যে বিশেষত কার্যকর যেমন ভেরিয়েবল ডেটাসেটগুলি প্রক্রিয়াজাতকরণ বা বাহ্যিক অবস্থার উপর ভিত্তি করে নির্দিষ্ট কাজগুলি সম্পাদন করা। এমন একটি ইটিএল পাইপলাইন কল্পনা করুন যেখানে ফাইলগুলি প্রতিদিনের পরিবর্তনের প্রক্রিয়া করতে পারে - এই পদ্ধতির অটোমেশনকে আরও সহজ করে তোলে। 🚀

প্রথম স্ক্রিপ্ট ব্যবহার করে পাইথনোপারেটর কার্য সম্পাদন এবং গতিশীলভাবে নির্ভরতা সেট করতে। এটি থেকে উপাদানগুলির তালিকা বের করে dag_run.conf, প্রয়োজন তখনই কাজগুলি তৈরি করা হয় তা নিশ্চিত করে। তালিকার প্রতিটি উপাদান একটি অনন্য কার্য হয়ে যায় এবং নির্ভরতাগুলি ক্রমানুসারে সেট করা হয়। দ্বিতীয় পদ্ধতির উত্তোলন টাস্কফ্লো এপিআই, যা সাজসজ্জার সাথে ডাগ সৃষ্টিকে সহজতর করে @ড্যাগ এবং @টাস্ক। এই পদ্ধতিটি ডিএজিটিকে আরও পঠনযোগ্য করে তোলে এবং ক্লিনার এক্সিকিউশন লজিক বজায় রাখে। এই পদ্ধতিগুলি নিশ্চিত করে যে ওয়ার্কফ্লোগুলি কোড পরিবর্তনের প্রয়োজন ছাড়াই বিভিন্ন কনফিগারেশনের সাথে খাপ খাইয়ে নিতে পারে।

উদাহরণস্বরূপ, এমন একটি দৃশ্য বিবেচনা করুন যেখানে একটি ই-বাণিজ্য সংস্থা ব্যাচগুলিতে অর্ডারগুলি প্রক্রিয়া করে। কিছু দিনের মধ্যে অন্যদের তুলনায় আরও জরুরি অর্ডার থাকতে পারে, বিভিন্ন টাস্ক সিকোয়েন্সের প্রয়োজন হয়। স্ট্যাটিক ডিএজি ব্যবহারের অর্থ প্রতিবার অগ্রাধিকার পরিবর্তনের সময় কোডটি সংশোধন করা। আমাদের গতিশীল ডিএজি পদ্ধতির সাথে, একটি বাহ্যিক সিস্টেম প্রক্রিয়াটিকে আরও দক্ষ করে তোলে, একটি নির্দিষ্ট টাস্ক সিকোয়েন্স দিয়ে ডিএজি ট্রিগার করতে পারে। আর একটি ব্যবহারের কেসটি ডেটা সায়েন্সে রয়েছে, যেখানে আগত ডেটা বিতরণের ভিত্তিতে মডেলগুলিকে পুনরায় প্রশিক্ষণের প্রয়োজন হতে পারে। গতিশীলভাবে প্রয়োজনীয় মডেল কনফিগারেশনগুলি পাস করে, কেবল প্রয়োজনীয় গণনাগুলি কার্যকর করা হয়, সময় এবং সংস্থানগুলি সংরক্ষণ করে। 🎯

সংক্ষেপে, এই স্ক্রিপ্টগুলি রানটাইম ইনপুটগুলির উপর ভিত্তি করে গতিশীলভাবে ডিএজি তৈরি করার জন্য একটি ভিত্তি সরবরাহ করে। লিভারিং দ্বারা এয়ারফ্লোর টাস্কফ্লো এপিআই বা traditional তিহ্যবাহী পাইথনোপারেটর পদ্ধতির, বিকাশকারীরা নমনীয়, মডুলার এবং দক্ষ কর্মপ্রবাহ তৈরি করতে পারে। এটি ম্যানুয়াল হস্তক্ষেপের প্রয়োজনীয়তা দূর করে এবং অন্যান্য অটোমেশন সিস্টেমগুলির সাথে বিরামবিহীন সংহতকরণের অনুমতি দেয়। গ্রাহকের অর্ডারগুলি প্রক্রিয়াজাতকরণ, ডেটা পাইপলাইন পরিচালনা করা, বা ক্লাউড ওয়ার্কফ্লোকে অর্কেস্ট্রেটিং, গতিশীল ডিএজিগুলি নির্দিষ্ট ব্যবসায়ের প্রয়োজন অনুসারে স্মার্ট অটোমেশন সক্ষম করে।

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 (ক্রস-যোগাযোগের বার্তা)। এক্সকোমগুলি কার্যগুলি ডেটা বিনিময় করতে দেয়, যার অর্থ একটি গতিশীলভাবে তৈরি টাস্ক সিকোয়েন্সটি পদক্ষেপের মধ্যে তথ্য পাস করতে পারে। উদাহরণস্বরূপ, একটি ইটিএল পাইপলাইনে, একটি প্রাক -প্রসেসিং টাস্ক প্রয়োজনীয় রূপান্তরগুলি নির্ধারণ করতে পারে এবং সেই বিবরণগুলি পরবর্তী কার্যগুলিতে পাস করতে পারে। এই পদ্ধতিটি সত্যিকারের ডেটা-চালিত ওয়ার্কফ্লোগুলিকে সক্ষম করে, যেখানে বাস্তবায়ন প্রবাহ রিয়েল-টাইম ইনপুটগুলির উপর ভিত্তি করে অভিযোজিত হয়, অটোমেশন ক্ষমতাগুলি উল্লেখযোগ্যভাবে বৃদ্ধি করে।