توليد تسلسل مهمة ديناميكي في تدفق الهواء باستخدام تكوين تشغيل DAG

فتح قوة تبعيات المهام الديناميكية في تدفق الهواء

Apache Airflow هي أداة تلقائية قوية لسير العمل ، لكن التعامل مع التبعيات الديناميكية قد يشعر أحيانًا بأنه يحل اللغز. عند تصميم الرسم البياني الموجه (DAG) ، قد تعمل تسلسل المهام المتشددين في حالات الاستخدام البسيطة ، ولكن ماذا لو كان هناك حاجة إلى تحديد الهيكل في وقت التشغيل؟ 🤔

تخيل أنك تعمل على خط أنابيب بيانات حيث تعتمد المهام التي سيتم تنفيذها على البيانات الواردة. على سبيل المثال ، معالجة مجموعات مختلفة من الملفات بناءً على التكوين اليومي أو تنفيذ التحولات المتغيرة بناءً على قاعدة عمل. في مثل هذه الحالات ، لن يقطعها DAG الثابت - أنت بحاجة إلى طريقة لتحديد التبعيات ديناميكيًا.

هذا بالضبط حيث تدفق الهواء dag_run.conf يمكن أن يكون مغير اللعبة. من خلال تمرير قاموس التكوين عند تشغيل DAG ، يمكنك إنشاء تسلسل المهام ديناميكيًا. ومع ذلك ، فإن تنفيذ هذا بطريقة منظمة يتطلب فهمًا عميقًا لنموذج تنفيذ تدفق الهواء.

في هذه المقالة ، سنستكشف كيفية إنشاء DAG ديناميكي حيث يتم تحديد تبعيات المهام في وقت التشغيل باستخدام dag_run.conf. إذا كنت تكافح من أجل تحقيق ذلك ولم تجد حلاً واضحًا ، فلا تقلق - فأنت لست وحدك! دعنا نقسمه خطوة بخطوة مع أمثلة عملية. 🚀

بناء dags الديناميكية مع تكوين وقت التشغيل في تدفق الهواء

Apache Airflow هي أداة قوية لتنظيم مهام سير العمل المعقدة ، ولكن قوته الحقيقية تكمن في مرونتها. توضح البرامج النصية المقدمة في وقت سابق كيفية إنشاء ملف داغ الديناميكي حيث يتم تحديد تبعيات المهام في وقت التشغيل باستخدام dag_run.conf. بدلاً من ترميز قائمة العناصر التي يجب معالجتها ، يسترجعها DAG ديناميكيًا عند تشغيلها ، مما يتيح المزيد من مهام سير العمل القابلة للتكيف. هذا مفيد بشكل خاص في سيناريوهات العالم الحقيقي ، مثل معالجة مجموعات البيانات المتغيرة أو تنفيذ مهام محددة بناءً على الظروف الخارجية. تخيل خط أنابيب ETL حيث تتغير الملفات اليومية - هذا النهج يجعل الأتمتة أسهل بكثير. 🚀

يستخدم البرنامج النصي الأول Pythonoperator لتنفيذ المهام وتعيين التبعيات ديناميكيا. يستخرج قائمة العناصر من dag_run.conf، ضمان إنشاء المهام فقط عند الحاجة. يصبح كل عنصر في القائمة مهمة فريدة ، ويتم تعيين التبعيات بشكل متتابع. النهج الثاني يعزز واجهة برمجة تطبيقات تدفق المهامالذي يبسط خلق DAG مع ديكور مثل dag و @مهمة. هذه الطريقة تجعل DAG أكثر قابلية للقراءة وتحافظ على منطق تنفيذ الأنظف. تضمن هذه الأساليب أن تتكيف مهام سير العمل مع تكوينات مختلفة دون طلب تغييرات رمز.

على سبيل المثال ، فكر في سيناريو حيث تقوم شركة التجارة الإلكترونية بمعالجة الطلبات على دفعات. قد يكون لدى بعض الأيام أوامر عاجلة أكثر من غيرها ، والتي تتطلب تسلسلات مهمة مختلفة. يعني استخدام DAG ثابت تعديل الكود في كل مرة يتغير فيها الأولويات. من خلال نهج DAG الديناميكي الخاص بنا ، يمكن للنظام الخارجي أن يؤدي إلى DAG بتسلسل مهام معين ، مما يجعل العملية أكثر كفاءة. هناك حالة استخدام أخرى في علم البيانات ، حيث قد تحتاج النماذج إلى إعادة التدريب بناءً على توزيعات البيانات الواردة. من خلال تمرير تكوينات النموذج المطلوبة ديناميكيًا ، يتم تنفيذ الحسابات اللازمة فقط ، وتوفير الوقت والموارد. 🎯

باختصار ، توفر هذه البرامج النصية أساسًا لتوليد DAGs ديناميكيًا استنادًا إلى مدخلات وقت التشغيل. عن طريق الاستفادة Airflow's Taskflow API أو نهج Pythonoperator التقليدي ، يمكن للمطورين إنشاء مهام سير عمل مرنة وعزيزة وفعالة. هذا يلغي الحاجة إلى التدخل اليدوي ويسمح بالتكامل السلس مع أنظمة الأتمتة الأخرى. سواء أكان معالجة طلبات العملاء أو إدارة خطوط أنابيب البيانات أو تدفقات سير العمل السحابية ، فإن 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()

تعزيز تسلسل المهام الديناميكية مع التنفيذ الشرطي في تدفق الهواء

ميزة واحدة قوية ولكن غالبا ما يتم تجاهلها في Apache Airflow هو التنفيذ المشروط ، والذي يمكن أن يزيد من تحسين مرونة تسلسل المهام الديناميكية. أثناء استرداد تبعيات المهام من dag_run.conf غالبًا ما تتطلب سيناريوهات العالم الحقيقي تنفيذ مهام معينة فقط بناءً على شروط محددة. على سبيل المثال ، قد تتطلب بعض مجموعات البيانات المعالجة المسبقة قبل التحليل ، بينما يمكن معالجة الآخرين مباشرة.

يمكن تنفيذ التنفيذ الشرطي في تدفق الهواء باستخدام BranchPythonOperator، والذي يحدد المهمة التالية التي يجب تنفيذها بناءً على المنطق المحدد مسبقًا. لنفترض أن لدينا DAG ديناميكي يقوم بمعالجة الملفات ، ولكن الملفات فقط فوق حجم معين تتطلب التحقق من الصحة. بدلاً من تنفيذ جميع المهام بالتتابع ، يمكننا تحديد مهام التشغيل بشكل ديناميكي ، وتحسين وقت التنفيذ وتقليل استخدام الموارد. يضمن هذا النهج تشغيل مهام سير العمل ذات الصلة فقط ، مما يجعل خطوط أنابيب البيانات أكثر كفاءة. 🚀

هناك طريقة أخرى لتعزيز DAGS الديناميكية من خلال دمجها XComs (رسائل التواصل المتبادل). تتيح XCOMs المهام لتبادل البيانات ، مما يعني أن تسلسل المهمة الذي تم إنشاؤه ديناميكيًا يمكن أن يمرر المعلومات بين الخطوات. على سبيل المثال ، في خط أنابيب ETL ، قد تحدد مهمة المعالجة المسبقة التحولات المطلوبة وتمرير هذه التفاصيل إلى المهام اللاحقة. تتيح هذه الطريقة سير العمل القائمة على البيانات حقًا ، حيث يتكيف تدفق التنفيذ على أساس المدخلات في الوقت الفعلي ، مما يزيد من قدرات الأتمتة بشكل كبير.