پارامتری کردن یک جریان‌کار از طریق فیلد Params در تعریف DAG – محتوای ویدئویی

یکی از مهم‌ترین قابلیت‌های ایرفلو برای ساخت جریان‌های کاری قابل استفاده مجدد و منعطف، پارامتری‌سازی (Parameterized Workflows) است. با کمک params می‌توانیم ورودی‌های DAG را در زمان اجرا مشخص کنیم، بدون اینکه نیاز به تغییر کد داشته باشیم. این ویژگی در سناریوهایی مثل تولید داده آزمایشی، پردازش داده‌های تاریخ خاص، یا کنترل مقیاس تسک‌ها بسیار حیاتی است.

اهداف یادگیری این بخش

• تعریف پارامترها در سطح DAG با استفاده از params و Param
• اعتبارسنجی ورودی‌ها با JSON Schema
• استفاده از پارامترها در داخل تسک‌ها
• اجرای موازی تسک‌ها با کمک Task Mapping
• ذخیره و مدیریت داده‌ها در MinIO
• آشنایی با فراخوانی Downstream DAG از طریق TriggerDagRunOperator

تعریف پارامترها در DAG

پارامترها در ایرفلو به‌صورت یک دیکشنری از Param تعریف می‌شوند. هر Param می‌تواند نوع داده، مقدار پیش‌فرض و محدودیت‌ها را مشخص کند.

params = {
    "max_tx_number": Param(2, type="integer", minimum=1),
    "start_date": Param(datetime.now(timezone.utc).isoformat(), type="string"),
    "end_date": Param(datetime.now(timezone.utc).isoformat(), type="string")
}

• max_tx_number: بیشترین تعداد تراکنش‌های تصادفی
• start_date و end_date: بازه زمانی تولید تراکنش‌ها

📌 نکته مهم برای دانشجویان: وقتی این DAG را از طریق Airflow UI تریگر می‌کنید، در همان لحظه یک فرم ورودی نمایش داده می‌شود که شامل سه فیلد است:

• برای max_tx_number یک فیلد عددی (Number Input) ظاهر می‌شود.
• برای start_date و end_date، چون از نوع string هستند، فیلد متنی (Text Input) نمایش داده می‌شود.

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

تولید تراکنش‌ها (Generate Transactions)

در این تسک از Faker برای ساخت داده‌های تصادفی استفاده می‌کنیم. تعداد و بازه‌ی زمانی تراکنش‌ها کاملاً توسط پارامترهای ورودی کنترل می‌شود.

tx_list = generate_transactions()

• ایجاد پروفایل کاربر و تراکنش
• تولید timestamp تصادفی در بازه داده‌شده
• بازگرداندن لیست تراکنش‌ها برای ادامه پردازش

نگهداری تراکنش‌ها در MinIO (Staging)

با استفاده از Task Mapping هر تراکنش به‌صورت مستقل سریالایز و به فرمت باینری (.bin) در MinIO ذخیره می‌شود.

staged_files = stage_transaction.expand(tx=tx_list)

نهایی‌سازی تراکنش‌ها (Finalize)

تبدیل فایل‌های باینری به JSON و حذف فایل‌های موقت:

final_files = finalize_transaction.expand(staged_file=staged_files)

تریگر کردن یک DAG پایین‌دست

final_files >> trigger_combine

نمونه ورودی برای اجرا

{
  "max_tx_number": 5,
  "start_date": "2025-09-14T00:00:00+00:00",
  "end_date": "2025-09-14T23:59:59+00:00"
}

جمع‌بندی نکات کلیدی

1. پارامتری‌سازی DAGها امکان پیکربندی پویا و متناسب با نیاز لحظه‌ای را فراهم می‌کند.
2. UI ایرفلو در هنگام تریگر یک فرم ورودی نمایش می‌دهد که بر اساس نوع داده، فیلد متنی یا عددی خواهد بود.
3. همین پارامترها در بخش Conf نیز به شکل JSON قابل ویرایش هستند.
4. با Task Mapping می‌توان تسک‌ها را به‌صورت موازی اجرا کرد.
5. MinIO به‌عنوان Object Storage برای نگهداری داده‌ها استفاده می‌شود.

محتوای ویدئویی

کارگاه عملی اول جلسه پنجم به اجرای یک دگ پارامتر دار میپردازد و مثالی که در بالا توضیح داده شده است را می توانید به صورت عملی در فیلم آموزشی زیر مشاهده کنید.

نکته : اگر فیلم در قسمت زیر قابل مشاهده نیست، مطمئن شوید که با آی پی ایران متصل شده اید یا یک اینترنت پروایدر دیگر را امتحان کنید .