🚫 چرا نباید مستقیم در فایل DAG کدنویسی کنیم؟

یکی از مهم‌ترین توصیه‌ها در بهترین شیوه‌های توسعه با Airflow این است که در فایل‌های DAG فقط باید تعریف تسک‌ها و جریان داده‌ها (Workflow Orchestration) انجام شود.
قرار دادن منطق پردازشی یا کدهای اجرایی خارج از Taskها می‌تواند مشکلات جدی ایجاد کند.

📌 مثالی از مشکل

در این مثال (Stock-Exchange-V1) قبل از تعریف DAG، یک کد Python برای تولید پروفایل جعلی با Faker نوشته شده:

import json
import uuid
from faker import Faker
from pathlib import Path
from datetime import date, datetime

# ---- Generate Random JSON Profile ----
fake = Faker()
profiles_dir = Path("/data/profiles")
profiles_dir.mkdir(parents=True, exist_ok=True)

profile = fake.simple_profile()

def json_serializer(obj):
    if isinstance(obj, (date, datetime)):
        return obj.isoformat()
    return str(obj)

file_path = profiles_dir / f"{uuid.uuid4()}.json"

with open(file_path, "w") as f:
    json.dump(profile, f, indent=2, default=json_serializer)

print(f"[INFO] Faker profile saved to {file_path}")
# --------------------------------------

این کد بیرون از DAG و خارج از هر Task قرار گرفته.

⚠️ چه اتفاقی می‌افتد؟

در Airflow 3، فایل‌های DAG توسط یک سرویس مستقل به نام DAG Processor Parse می‌شوند.

• قبلاً (تا Airflow 2) خود Scheduler هم DAGها را Parse می‌کرد.
• در Airflow 3 این دو بخش از هم جدا شدند:
  • DAG Processor: مسئول Parse و Import کردن فایل‌های DAG است.
  • Scheduler: فقط زمان‌بندی DAGها را انجام می‌دهد و از خروجی DAG Processor استفاده می‌کند.

👉 نتیجه این تغییر این است که هر بار DAG Processor فایل DAG را بررسی کند، تمام کدی که بیرون از Task نوشته شده اجرا می‌شود.

مشاهده عملی مشکل

اگر به پوشه /data/profiles نگاه کنید که در Docker Compose به پوشه data کنار پروژه مونت شده است، خواهید دید که حدود هر ۳۰ ثانیه یک فایل JSON جدید تولید می‌شود.
این دقیقاً همان Side Effect ناخواسته‌ای است که از قرار دادن کد خارج از Task به وجود می‌آید.

💡 چرا این کار اشتباه است؟

1. ایجاد Side Effect ناخواسته:
   هر بار که DAG Processor فایل DAG را Parse می‌کند، کد خارج از Task اجرا می‌شود و خروجی غیرمنتظره تولید می‌شود.
2. بار اضافی روی سیستم:
   اجرای مداوم کدها باعث کندی DAG Processor و حتی تأثیر روی Scheduler می‌شود.
3. پایداری پایین‌تر:
   یک خطای کوچک در کد خارج از Task می‌تواند مانع Parse شدن DAG شود و کل کلاستر را تحت تأثیر قرار دهد.
4. تست و نگهداری دشوار:
   منطق تجاری با orchestration قاطی می‌شود و کدها سخت‌تر تست و نگهداری می‌شوند.

✅ روش درست

۱. منطق پردازشی را در فایل جداگانه قرار دهید

مثلاً در مسیر plugins/utils/profile_generator.py:

import json, uuid
from faker import Faker
from pathlib import Path
from datetime import date, datetime

fake = Faker()
profiles_dir = Path("/data/profiles")
profiles_dir.mkdir(parents=True, exist_ok=True)

def generate_profile():
    profile = fake.simple_profile()
    file_path = profiles_dir / f"{uuid.uuid4()}.json"

    def json_serializer(obj):
        if isinstance(obj, (date, datetime)):
            return obj.isoformat()
        return str(obj)

    with open(file_path, "w") as f:
        json.dump(profile, f, indent=2, default=json_serializer)

    print(f"[INFO] Faker profile saved to {file_path}")

۲. در DAG فقط آن را به‌عنوان Task فراخوانی کنید

from airflow.operators.python import PythonOperator
from utils.profile_generator import generate_profile

task_generate_profile = PythonOperator(
    task_id="Generate-Profile",
    python_callable=generate_profile,
)

🎯 جمع‌بندی

• در Airflow 3، Parse فایل‌های DAG توسط DAG Processor انجام می‌شود.
• بنابراین هر کدی که خارج از Task باشد، همیشه و مداوم اجرا می‌شود.
• این کار باعث ایجاد Side Effect، مصرف منابع و مشکلات پایداری می‌شود.
• بهترین روش: منطق پردازشی در ماژول‌های جداگانه + استفاده از Operatorها در DAG.
• با نگاه کردن به پوشه /data/profiles در محیط Docker Compose می‌توانید اثر این اشتباه را به‌صورت عملی مشاهده کنید: فایل‌ها هر ۳۰ ثانیه تولید می‌شوند.