Data Engineering

خط لوله‌های داده مقیاس‌پذیر: تسلط بر DAGهای پویا و نگاشت وظایف در Airflow 2.0+

Oct 15, 2026 · 5 views
scalabilitypythondata engineeringETLAirflowDynamic DAGsTask Mapping

Airflow از دیرباز استاندارد غیررسمی برای هماهنگ‌سازی جریان‌های کاری داده بوده است. با این حال، با افزایش حجم داده‌ها و پراکنده‌تر شدن منابع، DAGهای ایستا اغلب با چالش‌های مقیاس‌پذیری و نگهداری‌پذیری مواجه می‌شوند. در Airflow 2.0 و نسخه‌های بعدی، معرفی نگاشت وظایف (Task Mapping) و بهبود تولید DAGهای پویا، ابزارهای محکمی برای ساخت خط لوله‌هایی فراهم می‌کند که بدون از دست دادن عملکرد یا شفافیت، با ساختارهای داده‌ای متغیر سازگار می‌شوند.

چرا DAGهای ایستا در مقیاس بزرگ شکست می‌خورند

DAGهای سنتی Airflow در زمان تجزیه (parse time) تعریف می‌شوند. اگر خط لوله شما نیاز به پردازش ۱۰۰ فایل داشته باشد، ممکن است ۱۰۰ وظیفه ایجاد کنید. این رویکرد منجر به موارد زیر می‌شود:

  • تورم کد: فایل‌های DAG عظیم با تعاریف تکراری وظایف.
  • زمان تجزیه طولانی: Airflow برای تجزیه DAGهای پیچیده زمان زیادی صرف می‌کند که باعث کند شدن زمان‌بند (scheduler) می‌شود.
  • عدم انعطاف‌پذیری: شما نمی‌توانید تعداد نامشخصی از فایل‌ها را بدون سخت‌کد کردن محدودیت‌ها یا استفاده از حلقه‌های پیچیده مدیریت کنید.

راه حل، گذار از «تعداد زیادی وظیفه» به «یک وظیفه نگاشت‌شده» است که چندین مورد را مدیریت می‌کند.

معرفی به نگاشت وظایف (Task Mapping)

نگاشت وظایف در Airflow 2.0 به شما امکان می‌دهد چندین نمونه وظیفه (task instance) را از یک تعریف وظیفه ایجاد کنید. وقتی از روش x.map() استفاده می‌کنید، Airflow برای هر عنصر در توالی قابل تکرار (لیست، تاپل یا ژنراتور) ارائه شده، یک اجرای جداگانه ایجاد می‌کند.

سینتکس پایه

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

with DAG(
    dag_id="task_mapping_example",
    schedule_interval="@daily",
    start_date=datetime(2023, 1, 1),
    catchup=False
) as dag:

    def process_file(file_name):
        print(f"Processing {file_name}")
        # Add your processing logic here
        return file_name

    # Define a list of files to process
    files = ["file1.csv", "file2.csv", "file3.csv", "file4.csv", "file5.csv"]

    # Map the task over the list of files
    map_task = PythonOperator(
        task_id="process_file_task",
        python_callable=process_file,
        op_kwargs={"file_name": files}  # Note: In Airflow 2.3+, use 'op_args' or 'op_kwargs' carefully
    )
    
    # Actually, the standard way for mapping a list of items to a function argument:
    # If process_file takes one arg, and we want to pass each item:
    
    # Corrected Implementation for Airflow 2.0+:
    def process_item(item):
        print(f"Processing: {item}")
        return item

    # We need to pass the list as an argument to map
    # Using op_args is not directly supported for mapping list elements in all versions.
    # Best practice: Use a wrapper or pass the list as op_kwargs and unpack?
    # Actually, Airflow 2.3+ recommends:
    
    mapped_task = PythonOperator(
        task_id="process_items",
        python_callable=process_item,
        op_args=files  # This is NOT how mapping works directly.
    )

اصلاح و بهترین روش: در Airflow، شما نمی‌توانید صرفاً یک لیست را به op_args بدهید تا آن را نگاشت کنید. باید از روش .x.map() روی نمونه اپراتور استفاده کنید. با این حال، PythonOperator به همان شیوه‌ای که BashOperator برای رشته‌های ساده دارد، روش مستقیم .x.map() ندارد. برای PythonOperator، الگوی رایج این است که کل لیست را به تابع بدهید و تکرار را درون آن مدیریت کنید، یا از یک زیرکلاس عمومی BaseOperator استفاده کنید، یا رایج‌تر، از op_kwargs استفاده کنید که مقدار آن لیست است، اما تابع قابل فراخوانی باید آن لیست را بپذیرد.

در واقع، محکم‌ترین راه برای نشان دادن نگاشت وظایف برای یک لیست موارد در یک PythonOperator این است که لیست را به عنوان آرگومان بدهید و بگذارید Airflow آن را تقسیم کند. اما صبر کنید، نگاشت PythonOperator با دادن هر عنصر لیست نگاشت‌شده به تابع قابل فراخوانی کار می‌کند.

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def process_file(file_name):
    print(f"Processing: {file_name}")
    return f"Processed {file_name}"

with DAG(
    dag_id="dynamic_processing",
    schedule_interval=None,
    start_date=datetime(2023, 1, 1),
) as dag:

    # Simulate a dynamic list of files (e.g., from a DB query or S3 bucket)
    files = [f"file_{i}.csv" for i in range(1, 101)]

    # Define the task
    process_task = PythonOperator(
        task_id="process_files",
        python_callable=process_file
    )

    # Map the task. This will create 100 task instances.
    # The list `files` is passed as the argument to map over.
    process_task.x.map(process_task, files)

نکته مهم: سینتکس .x.map() در بالا برای شفافیت کمی ساده‌سازی شده است. در عمل، اغلب از op_kwargs برای مشخص کردن استفاده می‌کنید که کدام پارامتر موارد نگاشت‌شده را دریافت می‌کند. اگر تابع شما چندین آرگومان می‌گیرد، باید اندیس آرگومانی که باید نگاشت شود را مشخص کنید.

def process_file(file_name, log_level="INFO"):
    print(f"[{log_level}] Processing: {file_name}")
    return f"Processed {file_name}"

# ... inside DAG ...
process_task = PythonOperator(
    task_id="process_files",
    python_callable=process_file,
    op_kwargs={"log_level": "DEBUG"}  # Fixed keyword argument
)

# Map over the first argument (file_name)
# index=0 refers to the first positional argument of the callable
process_task.x.map(process_task, files, index=0)

مدیریت داده‌های پویا: ژنراتورها و بافت (Context)

اغلب اوقات، شما لیست موارد را در زمان تعریف DAG نمی‌دانید. اینجاست که تولید DAGهای پویا و ژنراتورها می‌درخشند. به جای ساخت یک لیست عظیم در حافظه، از یک ژنراتور برای تولید موارد یکی‌یکی استفاده کنید. این برای خط لوله‌های مقیاس بزرگ که تعداد موارد می‌تواند در حد هزاران باشد، حیاتی است.

import boto3

def get_s3_files(bucket, prefix):
    s3 = boto3.client('s3')
    paginator = s3.get_paginator('list_objects_v2')
    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
        for obj in page.get('Contents', []):
            yield obj['Key']

# Inside DAG definition:
# Note: You cannot easily use a generator that hits S3 at parse time if the 
# bucket structure changes dynamically per run. 
# Best Practice: Use a Task to fetch the list, then map based on that output.

بهترین روش‌ها برای مقیاس‌پذیری

  1. محدود کردن تعداد وظایف: در حالی که نگاشت وظایف مشکل کد «تعداد زیادی وظیفه» را حل می‌کند، داشتن ۱۰,۰۰۰ نمونه وظیفه در یک اجرای واحد می‌تواند زمان‌بند و پایگاه داده Airflow را تحت فشار قرار دهد. اگر تعداد از ۵۰۰ تا ۱۰۰۰ بیشتر شد، در نظر گرفتن دسته‌بندی (batching) موارد را در نظر بگیرید.
  2. استفاده از پردازش دسته‌ای: اگر هر مورد کوچک است، موارد را در دسته‌ها گروه‌بندی کنید (مثلاً ۵۰ فایل در هر وظیفه) تا سربار (overhead) راه‌اندازی وظیفه و لاگ‌نویسی کاهش یابد.
  3. پایش عملکرد: تعداد نمونه‌های وظیفه تولید شده در هر اجرا را ردیابی کنید. از متریک‌هایی مانند airflow_dagrun_taskinstance_run_duration برای شناسایی گلوگاه‌ها استفاده کنید.
  4. مدیریت خطا: به طور پیش‌فرض، اگر یک وظیفه نگاشت‌شده شکست بخورد، کل اجرا به عنوان ناموفق علامت‌گذاری می‌شود. اگر نیاز دارید که شکست‌های فردی را بدون توقف کل خط لوله تحمل کنید، از partial_mapping=True (در دسترس در نسخه‌های جدیدتر) یا منطق سفارشی استفاده کنید.

نتیجه‌گیری

Airflow 2.0+ ویژگی‌های قدرتمندی برای ساخت خط لوله‌های داده‌ای فراهم می‌کند که با نیازهای کسب‌وکار شما مقیاس‌پذیر می‌شوند. با بهره‌گیری از نگاشت وظایف، می‌توانید از DAGهای شکننده و سخت‌کد شده به جریان‌های کاری انعطاف‌پذیر و پویا حرکت کنید که حجم‌های متغیر داده را به طور کارآمد مدیریت می‌کنند. به یاد داشته باشید که تعداد نمونه‌های وظیفه را پایش کنید و برای مجموعه‌های داده بسیار بزرگ، دسته‌بندی را در نظر بگیرید تا اطمینان حاصل کنید که زمان‌بند Airflow شما پاسخگو باقی می‌ماند. هنگام طراحی خط لوله بعدی خود، از خود بپرسید: «آیا می‌توان این را نگاشت کرد؟» اگر پاسخ مثبت است، از قدرت تولید پویای وظایف استقبال کنید.

Share:
Related Posts