DAG লেখা ও schedule

এটি Airflow-এর "interval-end" semantics — পরিকল্পিত design choice, ভুল নয়। বুঝতে গেলে batch processing-এর historical context দরকার।

মূল প্রশ্ন: "৫ মে-র data" কখন available?

• ৫ মে শেষ হওয়ার আগে — পুরো দিনের data নেই।
• ৫ মে শেষ হলে (অর্থাৎ ৬ মে ০০:০০-এ) — পুরো ৫ মে-র data যত হওয়ার তত।
• তাই "৫ মে process" মানে — ৬ মে চালান।

Airflow-এর rule:

• প্রতিটি DAG run একটি "interval" cover করে: [start, end)।
• Run trigger হয় interval-এর end-এ।
• execution_date (পুরাতন নাম) / data_interval_start = interval-এর start।
• {{ ds }} = data_interval_start-এর YYYY-MM-DD।

ফলাফল:

• "@daily" + start_date 2026-05-01 →
• প্রথম run: trigger ২ মে ০০:০০-এ, ds = 2026-05-01, ১ মে-র data process।
• দ্বিতীয় run: trigger ৩ মে ০০:০০, ds = 2026-05-02।

কেন এই design?

• Idempotency: WHERE date = '{{ ds }}' — সবসময় completed day-এর data।
• Backfill: পুরাতন interval পুনরায় run করা সহজ — ds পরিচিত।
• Hourly/sub-hourly: একই pattern — interval শেষে process।

Practical implication:

• "আজকের data" পেতে — {{ ds }} চলবে না (গতকাল process)।
• "Real-time" দরকার হলে — Airflow নয়, streaming।
• BI report-এ users কে বুঝান: "৫ মে-র data" = ৬ মে সকালে available।

Airflow ২.৪+ পরিবর্তন:

• data_interval_start ও data_interval_end introduced।
• {{ ds }} এখনো data_interval_start point করে।
• "Run At" UI-তে actual trigger time, "Logical Date" = ds।

Bangladesh fintech example: bKash daily settlement — ৫ মে-র সব transaction settle হবে ৬ মে ভোরে। DAG schedule "0 5 * * *" (UTC), ds = 2026-05-05, query "WHERE txn_date = '{{ ds }}'"। ৬ মে ১১:০০ AM BST-তে stakeholder report পায় — যা ৫ মে-র complete data।

মূল কথা: "data interval" mental model adopt করুন। তারিখ নিয়ে confusion আর হবে না।

এটি real production question। আগে DAG topology বুঝে নিই: ১০ task-এ অনেক branch থাকতে পারে।

Scenario analyze:

• Task 4 fail এবং Task 5-7 চলছে — মানে এরা Task 4-এর downstream নয় (parallel branch)।
• Task 4-এর downstream (যেমন Task 8) Task 4 success-এর জন্য wait করছিল।

Default behavior:

• Task 5-7 চলতে থাকবে — এরা Task 4-এর dependent না।
• Task 4 retry করবে (retries=3)।
• Task 4-এর downstream task (যেমন Task 8) upstream_failed state-এ skip হবে যদি retry-ও fail।

Trigger rules — দশটি option:

• all_success (default): সব upstream success হলে চলবে।
• all_failed: সব fail হলে — recovery task।
• one_success: যেকোনো ১টি success — branch merge।
• one_failed: যেকোনো ১টি fail — alert task।
• none_failed: কেউ fail না হলে (success বা skipped)।
• all_done: সব finish হলে (যেকোনো state) — cleanup task।

Production pattern: cleanup + alert

extract >> transform >> load
load >> cleanup       # trigger_rule="all_done"
load >> alert_failure # trigger_rule="one_failed"

"Stop everything on failure" pattern:

• fail_fast=True DAG-এ — একটি fail হলেই DAG-এর সব running task kill।
• Use case: critical sequential pipeline যেখানে partial completion বিপজ্জনক।
• Default off — overusage থেকে সাবধান।

"Continue despite failure" pattern:

• Downstream task-এ trigger_rule="all_done"।
• Failed branch-এর result null/skipped, কিন্তু others process।
• Use case: non-critical enrichment fail হলেও main pipeline চলবে।

Manual intervention:

• UI-তে failed task → "Clear" → re-queue।
• Downstream-ও clear করতে পারেন: cascade।
• "Mark Success" — manually fix data, তারপর mark।
• CLI: airflow tasks clear <dag_id> --task-ids extract।

Bangladesh-এর reality: Pathao analytics DAG-এ — ride extract fail হলেও driver settlement চালাতে চান। দু'টি independent branch বানান, alert task সব fail capture। main pipeline-এ partial result acceptable করুন।

মূল কথা: Task failure design-time-এ ভাবা — operational nightmare এড়ায়। trigger_rule + retry + alert-এর সঠিক combination = robust DAG।

XCom-এ ৫০০ MB — disaster। কেন এবং কী করণীয়, ব্যাখ্যা করি।

কী ঘটবে:

• Default XCom backend = metadata DB (PostgreSQL)।
• JSON serialize → ৫০০ MB blob → DB-তে INSERT।
• Failure cases:
• (ক) Postgres default max_allowed_packet — limit hit।
• (খ) Serialize-এ executor memory burst।
• (গ) Task B start-এর সময় ৫০০ MB pull → memory pressure।
• (ঘ) Metadata DB bloat — সব XCom retention period পর্যন্ত থাকে।
• (ঙ) Webserver UI slow — XCom view 500MB load করতে চাইবে।

Airflow guidance: XCom < ১ MB। এর বেশি হলেই warning sign।

Alternative ১: External storage + path passing

@task
def transform(date: str) -> str:
    df = ...  # 500 MB DataFrame
    path = f"s3://daraz-data/staging/{date}.parquet"
    df.to_parquet(path)
    return path   # ছোট string XCom

@task
def load(path: str):
    df = pd.read_parquet(path)
    # ...

সবচেয়ে common pattern। S3, GCS, local path — সব কাজ করে। URL/path = ছোট string।

Alternative ২: Custom XCom backend

• Airflow ২.০+ — BaseXCom.serialize_value override।
• বড় object S3-তে save, metadata DB-তে শুধু URI।
• Configuration: xcom_backend = my_module.S3XComBackend।
• Code transparent — task থেকে দেখতে normal XCom।

Alternative ৩: Single task

• Transform + load একই task-এ মার্জ — XCom লাগে না।
• Trade-off: visibility কম (২ task-এর বদলে ১)।
• Retry granularity কম।

Alternative ৪: Spark/cluster compute

• ৫০০ MB pandas-এ — সম্ভবত Spark/Dask-এ যাওয়ার সময়।
• Airflow শুধু Spark job submit করে — data Spark cluster-এ থাকে।
• XCom-এ শুধু job_id বা output path।

সাধারণ ভুল:

• return df.to_dict() — XCom-এ python dict, JSON serializable, কিন্তু still huge।
• List of dict — Postgres TEXT field overflow।
• Image bytes XCom-এ — totally wrong।

Best practice rule:

• XCom = metadata, control flow, references (URI, ID, count)।
• Real data = external storage।
• "data lake first" mindset।

মূল কথা: Airflow orchestrator — data mover না। Heavy data work systems (Spark, dbt, Snowflake)-এ; Airflow শুধু coordinate।

এটি real-world API integration challenge — multiple valid approach আছে, trade-off-ভিত্তিক।

Option ১: Simple retry

@task(retries=5, retry_delay=timedelta(minutes=1))
def fetch_locations(ds_nodash):
    response = requests.get(API_URL, timeout=30)
    response.raise_for_status()
    return response.json()

• সরল, কাজ করে।
• ৫ × ১ min = ৫ min retry window।
• API down হলে — alert ৫ min পর।
• Drawback: প্রতি retry-এ task slot occupy।

Option ২: Exponential backoff

retries=5,
retry_exponential_backoff=True,
retry_delay=timedelta(seconds=30),
max_retry_delay=timedelta(minutes=10),

• Wait: 30s, 60s, 120s, 240s, 480s।
• Transient failure-এ ভাল — API recovering গেলে দ্রুত পেয়ে যাবে।
• Long outage-এ — ১৫ মিনিট পর্যন্ত wait।

Option ৩: HttpSensor + reschedule mode

wait_api = HttpSensor(
    task_id="wait_api",
    http_conn_id="pathao_api",
    endpoint="/health",
    poke_interval=30,
    timeout=600,  # 10 min max wait
    mode="reschedule",  # slot ছেড়ে দেয়
)
wait_api >> fetch_locations

• Sensor ৩০ সেকেন্ডে একবার health check।
• mode="reschedule" — wait time-এ worker slot free।
• API ready হলেই fetch।

Option ৪: Increase task timeout

@task(execution_timeout=timedelta(minutes=10))
def fetch_locations(...):
    # API ধৈর্যের সাথে wait
    response = requests.get(API_URL, timeout=300)

• সবচেয়ে সরল।
• API connection long-poll করে।
• Drawback: worker slot block।

Option ৫: Circuit breaker pattern

• @task এ tenacity library দিয়ে complex retry logic।
• API consecutive ৫ fail হলে — temporarily skip।
• Production-grade resilience।

Option ৬: Async / decoupling

• API webhook-এ event push, queue (SQS/Kafka)।
• Airflow polling বদলে event-driven।
• Architecture change — large scale-এ valuable।

কোনটা বাছবেন:

• Pathao 15-min cycle: Option ২ (exponential) + Option ৪ (timeout) combine।
• Reasoning: 15 min = tight schedule। দীর্ঘ wait lattice cycle miss করবে।
• SLA decision: ১২ মিনিটে শেষ না হলে — current cycle drop, পরের cycle-এ catch up।

Monitoring critical:

• Task duration metric — কখন বেশি wait?
• Retry count — API quality indicator।
• Failure rate — alert threshold।

মূল কথা: External API integration-এ retry/timeout/sensor combine করুন। কিন্তু ১৫ min batch-এ ১০ min wait — design pattern পরিবর্তন (event-driven) ভাবার সময়।