Apache Airflow পরিচিতি

Executor — Airflow-এর সবচেয়ে গুরুত্বপূর্ণ deployment decision। Workload, scale, infrastructure সব এর উপর নির্ভর করে।

SequentialExecutor:

• একসাথে শুধু ১টি task চলে।
• SQLite metadata-এর সাথে default।
• Use: শুধু local development, "hello world" শেখা।
• Production-এ কখনো না।

LocalExecutor:

• Scheduler ও worker একই machine-এ। parallel task সম্ভব।
• PostgreSQL/MySQL metadata দরকার।
• Single VM-এ ১০-৫০টি concurrent task। সহজ setup।
• Use: ছোট startup (১-২০ DAG), single team, ৪-৮ vCPU machine।
• Bangladesh: ছোট fintech-এর internal reporting — ১টি ৪ vCPU instance যথেষ্ট।

CeleryExecutor:

• Worker pool — আলাদা machine-এ। Redis/RabbitMQ message broker।
• Horizontal scaling — worker সংখ্যা বাড়ান।
• Setup complex: Redis, Celery, multiple worker process।
• Use: medium scale (৫০-৫০০ DAG, ১০০-১,০০০ concurrent task)।
• Bangladesh: bKash, Daraz, Grameenphone — typical।
• Drawback: worker pool fixed size, idle resource waste।

KubernetesExecutor:

• প্রতিটি task একটি new pod-এ।
• Auto-scaling — task নেই, pod নেই, খরচ নেই।
• Per-task isolation — dependencies কাছাকাছি থাকে না।
• K8s cluster দরকার, complexity high।
• Use: large scale (১,০০০+ DAG), variable workload, cloud-native team।
• Bangladesh: বড় bank, telecom, AI startup-এর serious data team।

হাইব্রিড — CeleryKubernetesExecutor:

• Light task: Celery (fast)। Heavy/special: K8s (isolated)।
• ২.০+ থেকে available।

Bangladesh-এর scale-ভিত্তিক recommendation:

• Solo data engineer / startup: LocalExecutor
• Daraz/Pathao scale: CeleryExecutor + managed Postgres
• bKash/GP scale: KubernetesExecutor on EKS/AKS/GKE
• Don't want ops: MWAA (AWS) বা Cloud Composer (GCP) — managed

মূল কথা: Executor decision = trade-off ops complexity বনাম scaling। ছোট থেকে শুরু, প্রয়োজনে migrate করুন।

Idempotency — Latin "idem" (একই) + "potent" (ক্ষমতা)। মানে: একই কাজ ১ বার বা ১০০ বার চালালেও result একই। Airflow-এ এটি pipeline reliability-র মূল ভিত্তি।

কেন critical:

• Retry: task fail হলে Airflow auto-retry। প্রতিটি attempt-এ একই side-effect হলে — duplicate data!
• Backfill: পুরাতন তারিখ পুনরায় চালাতে চান। Idempotent না হলে — সংখ্যা ভুল।
• Manual re-run: bug fix করে re-run — পুরাতন output overwrite হবে নাকি double-count?

Non-idempotent উদাহরণ (bKash style):

def increment_counter():
    db.execute("UPDATE stats SET total = total + 1")
# Run 3 বার = total ৩ বার বাড়বে। ভুল!

Idempotent ভার্সন:

def set_counter_for_date(date):
    db.execute(
        "INSERT INTO stats (date, total) VALUES (%s, %s) "
        "ON CONFLICT (date) DO UPDATE SET total = EXCLUDED.total",
        (date, calculate_total(date))
    )
# Run ১০০ বার = একই row, একই value

Idempotency patterns:

• Date-partitioned writes: partitionBy("date") — re-run শুধু ওই partition overwrite।
• UPSERT (MERGE): SQL-এ INSERT OR UPDATE — duplicate handle।
• Delete-before-insert: DELETE WHERE date='X'; INSERT ... — full replace।
• Content-addressable storage: hash-ভিত্তিক file name — একই content একই file।

Airflow-specific helpers:

• execution_date / logical_date: task-এর "logical" তারিখ। প্রকৃত run time নয়। Backfill-এ এটি pass হয়।
• Macro: {{ ds }} = execution date — query-তে inject। প্রতি run-এ আলাদা।
• External ID: Spark job-এ Airflow run_id pass — duplicate detect।

সাধারণ ভুল:

• now() বা CURRENT_DATE ব্যবহার — backfill ভাঙবে। {{ ds }} macro use করুন।
• API call without idempotency key — payment double-charge!
• Append-only logs without dedup — multi-run-এ duplicate row।

Bangladesh fintech context: bKash-এ একটি payment confirmation API non-idempotent হলে — retry-তে ২ বার customer-এর account debit। Idempotency key (transaction_id) প্রতিটি API call-এ critical।

মূল কথা: Idempotent না হলে — Airflow চালাবেন না। প্রতিটি task design-এর সময় ভাবুন: "এটি ৫ বার চালালে result-এ পার্থক্য হবে কি?" না হলে ✓।

Airflow-এর scheduler প্রতি কয়েক সেকেন্ডে সব DAG file (Python) parse করে — শুধু schema বুঝতে নয়, পুরো Python execute করে! এটি ভয়াবহ slow হতে পারে।

Parsing process:

1. Scheduler ~/airflow/dags/ ফোল্ডার scan।
2. প্রতিটি .py file Python interpreter-এ load।
3. DAG object ও task definition extract।
4. Metadata DB-তে current state-এর সাথে compare।
5. Need-to-run task queue-তে push।

সমস্যা যেখানে:

• Top-level code execution: DAG file-এ requests.get(...), DB query — প্রতিবার parsing-এ চলে! ১,০০০ DAG × ১ সেকেন্ড = ১৬+ মিনিট/cycle।
• Heavy import: import pandas as pd top-level — slow।
• Dynamic DAG generation: for-loop-এ ১০০ DAG তৈরি করলে — সবগুলো প্রতি parse-এ।

Optimization কৌশল:

• Top-level কোড সরান: Variable, connection লোড task function-এর ভেতরে। DAG file শুধু structure।
• parse interval বাড়ান: min_file_process_interval = 30 — ৩০ সেকেন্ডে একবার।
• parsing parallelism: parsing_processes = 4 — multi-core ব্যবহার।
• DAG file split: ছোট ছোট ফোল্ডার-এ ভাগ। .airflowignore দিয়ে কিছু skip।

Airflow ২.০+ DAG Serialization:

• Webserver আর DAG file parse করে না — DB থেকে JSON serialized version পড়ে।
• UI অনেক দ্রুত।
• default-এ on।

Airflow ২.৩+ Dynamic Task Mapping:

• For-loop দিয়ে DAG তৈরি না করে — .expand() ব্যবহার করুন।
• Runtime-এ task সংখ্যা decide।
• Parsing time-এ scale handle।

Diagnostic command:

# প্রতিটি DAG file-এর parsing time
airflow dags list-import-errors
airflow dags list -o yaml | grep duration

# Webserver-এ Browse → DAG Parsing
# > ১ সেকেন্ড হলে problem

Production scale (Daraz/বড় telecom):

• ৫০০+ DAG: parse time monitor। ৩০s অতিক্রম হলে alert।
• Heavy DAG আলাদা scheduler instance-এ (Airflow ২.০+ multi-scheduler)।
• DAG বানানো-এর CI/CD-তে static check (linter)।

মূল কথা: "DAG file = pure structure declaration" — এই principle মাথায় রাখুন। Side-effect function-এর ভেতরে। তাহলে scheduler smooth।

এটি প্রকৃত production scenario। ভুল migration = ৬ মাস কাজ ও ক্ষুব্ধ stakeholder। সঠিক migration = ৩ মাসে stable Airflow।

Phase 1: Inventory ও audit (১-২ সপ্তাহ)

• সব ৩০ cron job তালিকা: schedule, dependency, owner, criticality।
• প্রতিটির runtime, fail rate, last-failure (যদি jana থাকে)।
• "Hidden dependency" খুঁজুন — script-এ অন্য script-কে call?
• Stakeholder map: কোন report কে দেখে?

Phase 2: Airflow setup (২-৩ সপ্তাহ)

• Managed Airflow recommend (MWAA on AWS, Composer on GCP)। In-house ৬ মাস delay।
• PostgreSQL metadata DB।
• CeleryExecutor (GP scale-এ), ৫-১০ worker দিয়ে শুরু।
• SSO integration (LDAP/Azure AD) — security টিম-এর জন্য।
• Slack/email alert wiring।

Phase 3: Pilot migration (৪ সপ্তাহ)

• ৩-৫টি simple, low-risk DAG বাছুন।
• Cron-এর সাথে parallel চালান (dual write)।
• ২ সপ্তাহ result compare — সব মিলছে?
• Confidence build করুন।
• Decommission cron version।

Phase 4: Bulk migration (৬-৮ সপ্তাহ)

• প্রতি সপ্তাহে ৫-৭ DAG migrate।
• Pattern library তৈরি — common operator wrap।
• Idempotency add — retry safe করুন।
• Code review process — junior পেরিয়ে যান না।
• Critical DAG (CDR processing, billing) — শেষে, parallel run বেশি সময়।

Phase 5: Stabilize ও improve (চলমান)

• SLA monitor — কোন DAG সময়মতো শেষ না?
• Cost analysis — worker right-size।
• Documentation — runbook প্রতিটি DAG-এর।
• On-call rotation।

মূল কথা: "Big bang" migration-এ ভাঙবে। Iterative, parallel-run, low-risk first — এই pattern অনুসরণ করুন। Airflow আপনার team-এর জন্য — না এর বিপরীত।