Build production Airflow DAGs, master idempotent and safe-rerun pipeline design, and progress through ingestion patterns from full load through incremental watermarks to advanced CDC with Debezium.
It is 3 AM at a company like Swiggy. A cron job that pulls yesterday's orders from Postgres into the warehouse silently failed four hours ago because the database connection timed out. Nobody knows. No retry happened. No alert fired. By the time an analyst opens the morning dashboard and asks why GMV looks 40% lower than yesterday, the team has already lost half a day chasing a problem that a five-second retry would have fixed on its own. This is the problem every data engineer eventually runs into with plain scripts and cron. A cron job knows how to start a script at a fixed time. It does not know if the script actually finished. It does not know if the script failed halfway through. It does not know that task B depends on task A completing successfully first. It has no history, no dashboard, and no way to alert anyone. An **orchestrator** is a system that schedules, runs, monitors, and retries pipeline tasks, and keeps a full history of every run so you can see exactly what happened and when. **Apache Airflow** is the industry-standard orchestrator for batch data pipelines, used at Flipkart, Swiggy, and most Indian data teams running anything beyond a handful of scripts. > 📌 **Remember:** Airflow does not move or transform your data itself. It tells your code when to run, watches whether it succeeded, and retries or alerts when it does not. The actual extraction, transformation, and loading logic still lives in your Python functions, SQL, or Spark jobs. ### What breaks without an orchestrator * No retries - a transient network blip kills the whole pipeline for the day * No dependency management - task B might start before task A's output exists * No visibility - nobody knows a pipeline failed until a stakeholder complains * No history - you cannot answer "did yesterday's load actually run, and when" * No backfilling - reprocessing three months of historical data means writing a new one-off script ### Why Airflow specifically Airflow represents a pipeline as a **DAG** - a graph of tasks with defined dependencies, written in Python. That Python definition is version-controlled, testable, and reviewable in a pull request, the same way application code is. This is a large part of why Airflow became the default: a data engineer already comfortable with Python does not need to learn a separate proprietary scheduling language. > 💡 **Practice:** Before continuing, write down (on paper or in a text file) the last time a script you wrote failed silently, or ran successfully but produced wrong output because of a race condition with another script. What would have caught it sooner - a retry, an alert, or a dependency check? Keep that answer in mind as you read the rest of this module. **Concept check:** What is the difference between what cron does and what an orchestrator like Airflow does? If you cannot answer this in one sentence, reread the section above before moving on. ---
Before writing a single DAG, it helps to know what is actually running when you use Airflow, because most beginner confusion (a task that "should have run" but did not) traces back to a misunderstanding of these pieces. The diagram below separates the two things that actually execute your tasks (Scheduler, Executor, Worker) from the two things that just read and write state (Webserver, Metadata DB) - the Webserver never sits in the execution path, it only displays what the database already says happened. EXECUTION PATH: Scheduler -> Executor -> Worker(s) -> runs your task code CONTROL / STATE PLANE: Scheduler <-> Metadata DB <-> Webserver (Airflow UI) Triggerer (separate process): handles deferrable sensors/operators while they wait, without occupying a Worker slot > **Note:** A typical Airflow deployment includes these core components, though the exact execution architecture depends on the executor and deployment model you choose. The Scheduler is the brain - it reads your DAG files, decides what should run based on the schedule, and hands tasks to the Executor, which decides how and where they actually run. The Webserver only reads from and writes to the Metadata Database - it is not part of the execution path, so the UI going down does not stop your pipelines. * **Scheduler** - the process that continuously reads your DAG files, evaluates schedules, and decides which task instances are ready to run * **Webserver** - serves the Airflow UI where you view DAG status, read logs, and manually trigger runs * **Metadata Database** - stores every DAG run, task state, and log reference; this is the actual source of truth, not the UI * **Executor** - determines how and where task code actually runs; see the next section * **Worker(s)** - the processes, used by distributed executors, that execute the actual task code (a PythonOperator function, a Bash command, and so on) * **Triggerer** (in newer Airflow versions) - a separate process that efficiently handles deferrable operators and sensors without occupying a full worker slot while waiting ### Understanding the Executor The **Executor** is the piece that decides how and where your task code actually runs once the Scheduler says a task is ready. This matters because it directly affects assumptions you can safely make about your task code - most importantly, whether two tasks in the same DAG run are guaranteed to execute on the same machine. * **LocalExecutor** - runs tasks as local processes on the same machine as the Scheduler; common for small deployments and local development * **CeleryExecutor** - distributes tasks across a pool of separate Worker machines; common in production at scale * **KubernetesExecutor** - launches a new, isolated pod per task on a Kubernetes cluster > 📌 **Remember:** With CeleryExecutor or KubernetesExecutor, two tasks in the same DAG run can execute on entirely different machines with entirely separate local filesystems. Any pipeline design that assumes "task A wrote a file, so task B can just read it from the same local path" is only safe under LocalExecutor - it silently breaks the moment a team moves to a distributed executor. Always pass data between tasks through shared storage (S3, a database, a shared volume), not a local file path. ### The three core concepts * **DAG (Directed Acyclic Graph)** - your pipeline defined as a graph of tasks with dependencies; "acyclic" means no task can depend on itself, directly or through a loop * **Task** - a single unit of work inside a DAG (extract, transform, load, validate) defined using an Operator * **Connection** - a stored, reusable credential set for a database, API, or cloud service, configured once in the Airflow UI or via environment variables, never hardcoded in DAG code > 🔴 **Common Mistake:** Assuming the Airflow UI is where task state lives. If the Webserver is down, your DAGs keep running exactly on schedule - the Scheduler and Workers do not depend on the Webserver at all. Beginners sometimes panic when the UI is slow or unreachable and assume pipelines have stopped, when in fact only the dashboard is affected. **Concept check:** If the Airflow Webserver crashes at 2 AM, does your nightly DAG still run on schedule? Why or why not? ---
Every DAG file is a Python script that Airflow's Scheduler reads and parses. Getting the DAG definition right - the arguments that control scheduling behavior - matters more than most beginners expect, because a wrong setting here can trigger hundreds of unwanted historical runs the moment you deploy. > **Note:** Newer Airflow versions accept `schedule` as the DAG argument name (`schedule="0 2 * * *"`), while `schedule_interval` is the older name for the same thing and still works widely in the wild. This module uses `schedule_interval` in most examples since it remains extremely common in production codebases and documentation, but check which your specific Airflow version and team convention expects, and prefer `schedule` on a fresh Airflow installation if it is available. ```python from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator from airflow.operators.bash import BashOperator def extract_orders(**context): """Pull the previous day's Swiggy order data using the logical date Airflow passes in via context - never use datetime.now() here, since that breaks backfilling for past dates.""" logical_date = context["ds"] # e.g. "2026-08-15" print(f"Extracting orders for {logical_date}") # actual extraction logic goes here default_args = { "owner": "data-engineering", "retries": 3, "retry_delay": timedelta(minutes=5), } with DAG( dag_id="swiggy_daily_orders_etl", schedule_interval="0 2 * * *", # 02:00 UTC every day start_date=datetime(2026, 1, 1), catchup=False, # do NOT backfill on deploy default_args=default_args, tags=["orders", "daily"], ) as dag: extract = PythonOperator( task_id="extract_orders", python_callable=extract_orders, ) transform = BashOperator( task_id="transform_orders", bash_command="python /opt/scripts/transform_orders.py {{ ds }}", ) load = BashOperator( task_id="load_orders", bash_command="python /opt/scripts/load_orders.py {{ ds }}", ) extract >> transform >> load ``` > **Note:** `{{ ds }}` is a Jinja template variable Airflow substitutes with the DAG run's **logical date** (for example `2026-08-15`) at execution time. "Logical date" is the term worth learning over the older "execution date" - it is deliberately not the wall-clock time the task actually ran. A DAG scheduled `@daily` that runs at 02:00 UTC on August 16th has a logical date of August 15th, because that run is processing the **data interval** for August 15th (the full day that just completed). Airflow exposes this data interval directly as `context["data_interval_start"]` and `context["data_interval_end"]`, which is more precise than `ds` alone when your pipeline cares about the exact window of data being processed, not just a single date label. Using the logical date (via `{{ ds }}` or the data interval) instead of hardcoding today's date is what makes a DAG safely backfillable - rerunning for any past date processes that date's data, not whatever `datetime.now()` happens to return when the task executes. ### Operators - the building blocks of tasks * **PythonOperator** - runs a Python function as a task; the most flexible and most commonly used operator * **BashOperator** - runs a shell command or script; useful for calling existing CLI tools or scripts in other languages * **Sensors** (a special operator subclass) - wait for a condition before letting downstream tasks proceed (see the next section) * Cloud-specific operators exist for most AWS/GCP services (S3, Glue, BigQuery), but the two above cover most beginner and intermediate pipelines ### The TaskFlow API - a cleaner way to write Python-heavy DAGs Everything above uses `PythonOperator` explicitly, which is the classic and still entirely valid way to write Airflow DAGs. Modern Airflow also offers the **TaskFlow API**, which uses a `@task` decorator to turn a plain Python function into a task, and lets you pass data between tasks by simply calling one function's output as another's input - Airflow handles the XCom push and pull underneath automatically. ```python from airflow.decorators import dag, task from datetime import datetime @dag( dag_id="swiggy_daily_orders_etl_taskflow", schedule_interval="0 2 * * *", start_date=datetime(2026, 1, 1), catchup=False, tags=["orders", "daily", "taskflow"], ) def orders_pipeline(): @task def extract(**context): logical_date = context["ds"] print(f"Extracting orders for {logical_date}") return {"row_count": 12_835} # returned value is auto-pushed to XCom @task def transform(extracted: dict): # extracted is auto-pulled from XCom - no manual xcom_pull needed print(f"Transforming {extracted['row_count']} rows") return extracted @task def load(transformed: dict): print(f"Loading {transformed['row_count']} rows") load(transform(extract())) orders_pipeline() ``` > **Note:** This produces the exact same DAG structure as the `PythonOperator` version - three tasks, `extract >> transform >> load` - but the dependency and data-passing wiring is implicit in how you call the functions, rather than written out with `>>` and manual `xcom_push`/`xcom_pull` calls. Both styles are correct Airflow and you will see both in production codebases; `PythonOperator` is more explicit and often clearer for beginners learning what XCom actually does under the hood (which is why this module leads with it), while TaskFlow reduces boilerplate once you already understand that mechanism. Non-Python tasks (BashOperator, sensors, cloud operators) still look the same in either style. ### Task dependencies The `>>` operator declares "runs after." `extract >> transform >> load` means transform will not start until extract succeeds, and load will not start until transform succeeds. You can also write `load << transform << extract`, though `>>` reads more naturally left to right. > 📌 **Engineering Decision:** Reach for Airflow, not a simple cron job, the moment any of these are true: a task depends on another task's success, you need automatic retries with backoff, you need to backfill historical date ranges, or more than one person needs visibility into whether a pipeline ran. If you have one standalone script with no dependencies and nobody besides you needs to know if it failed, a cron job with a basic exit-code check is genuinely simpler and Airflow would be overkill. Most pipelines that start as "just one cron job" grow a second and third step within weeks - that is usually the point to migrate. > 💡 **Practice:** Take the DAG above and add a fourth task, `validate_load`, that runs after `load` and checks that the target table has more than zero rows for the given date, raising an exception if not. Wire it in with `>>` so it only runs after `load` succeeds. **Concept check:** Why does `extract_orders` read the execution date from `context["ds"]` instead of calling `datetime.now()`? ---
Tasks in a DAG frequently need to share small values - a row count, a file path, a boolean flag - with each other. Airflow has a specific, limited-purpose mechanism for this called XCom, and knowing its limits is more important than knowing its syntax. * **XCom ("cross-communication")** - a small key-value store in the Metadata Database that lets one task push a value and another task pull it * **Variables** - configuration values (like an API base URL or a threshold) stored once in Airflow and reused across many DAGs, changeable without redeploying code * **Connections** - stored credentials for databases and APIs, referenced by a connection ID in your task code rather than typed directly into it ```python from airflow.operators.python import PythonOperator def check_row_count(**context): """Push the extracted row count so downstream tasks can use it.""" row_count = 12_835 # result of a real query in production context["ti"].xcom_push(key="row_count", value=row_count) def alert_if_low(**context): """Pull the row count pushed by the upstream task.""" row_count = context["ti"].xcom_pull( task_ids="check_row_count", key="row_count" ) if row_count < 1000: raise ValueError(f"Row count too low: {row_count}") ``` > 🔴 **Common Mistake:** Using XCom to pass an entire DataFrame or file's worth of data between tasks. XCom values are stored as rows in the Airflow metadata database, which is not built to hold megabytes of data - doing this slows down the whole Airflow instance and can crash the scheduler. Pass a **path** through XCom (an S3 key, a local file path) and let the downstream task read the actual data from there. **Concept check:** You need to pass a 500 MB Parquet file's contents from your extract task to your transform task. Should you use XCom directly? What should you do instead? ---
Not every dependency lives inside a single DAG. Sometimes a pipeline needs to wait for a file that another team drops into S3, or wait for a completely separate DAG to finish before it starts. This is what **sensors** are for. DAG A: upstream_ingestion +------------------+ | extract -> load | +--------+---------+ | (success) v DAG B: downstream_reporting +----------------------------+ | ExternalTaskSensor | | (waits for DAG A to finish) | +--------------+-------------+ | v build_report * **FileSensor** - waits for a file to appear on a filesystem accessible to the Airflow task; it does not check S3 or other object storage * **S3KeySensor** - the S3-specific equivalent; waits for an object/key to appear in an S3 bucket, which is what you actually want for the "wait for a file dropped in S3" scenario, not FileSensor * **ExternalTaskSensor** - waits for a specific task in a different DAG to reach a successful state before continuing * Sensors have a `poke_interval` (how often to check) and a `timeout` (how long to wait before failing) - always set a timeout, or a sensor can hang indefinitely and quietly consume a worker slot ```python from airflow.sensors.external_task import ExternalTaskSensor wait_for_ingestion = ExternalTaskSensor( task_id="wait_for_ingestion", external_dag_id="swiggy_daily_orders_etl", external_task_id="load_orders", timeout=3600, # give up after 1 hour poke_interval=60, # check every 60 seconds mode="reschedule", # free the worker slot between checks ) ``` > 💡 **Tip:** Set `mode="reschedule"` on long-running sensors instead of the default `mode="poke"`. Poke mode holds a worker slot the entire time it waits, which can starve other tasks in a busy Airflow instance. Reschedule mode releases the slot between checks and only reclaims one when it is time to check again. There is a third option worth knowing about: some operators and sensors support **deferring**, which hands the entire wait off to the Triggerer process introduced earlier, freeing the worker slot completely rather than just releasing it between polls. The progression is poke (worker occupied the whole time) -> reschedule (worker released between checks, reclaimed to poll) -> deferrable (worker released entirely, Triggerer handles the wait) - each step frees up more capacity for other tasks to run. **Concept check:** Why does a sensor need a `timeout`, and what happens to a worker slot if you forget to set one in `poke` mode? ---
A pipeline that runs correctly once is easy. A pipeline that runs correctly every single day, including the days something upstream breaks, is the actual job. Three properties separate a reliable pipeline from a fragile one: retries, idempotency, and alerting. ### Retries ```python default_args = { "retries": 3, "retry_delay": timedelta(minutes=5), } ``` This configuration means a failed task gets three *additional* attempts after its first try, for a maximum of four attempts total, waiting five minutes between each one. Most production failures - a database connection timeout, a rate-limited API call, a brief network partition - resolve themselves within a retry or two. Airflow's built-in retry logic turns what would have been a 3 AM page into a non-event. > 🔴 **Common Mistake:** Reading `"retries": 3` as "the task runs three times total." It actually means one initial attempt plus three retries, so up to four attempts run before the task is marked failed. This trips people up when they are counting log entries or trying to reason about total wall-clock time before a final failure alert fires. Retries protect against a task *failing*, but they do nothing if a task simply hangs and never finishes or fails on its own - a query stuck waiting on a lock, an API call with no response. For that, set an explicit **task timeout**: ```python default_args = { "retries": 3, "retry_delay": timedelta(minutes=5), "execution_timeout": timedelta(minutes=30), } ``` `execution_timeout` forces Airflow to mark a task as failed (triggering retries, if configured) once it has been running longer than the given duration. Without it, a stuck task can occupy a worker slot indefinitely, silently blocking every other task waiting behind it - retrying is not the same as timing out, and production pipelines need both. ### Idempotency **Idempotent** means running a task twice produces the exact same result as running it once. This property is what makes retries and backfills safe rather than dangerous. Consider a task that appends new rows to a table every time it runs: ```sql -- NOT idempotent - running this twice doubles every row INSERT INTO silver.orders SELECT * FROM bronze.orders WHERE ingest_date = '2026-08-15'; ``` If this task fails halfway through and Airflow retries it, or if someone accidentally triggers a manual rerun, the table now has duplicate rows for that date - and nobody notices until a revenue report is double-counted. ```sql -- Idempotent - reruns produce the same end state every time DELETE FROM silver.orders WHERE ingest_date = '2026-08-15'; INSERT INTO silver.orders SELECT * FROM bronze.orders WHERE ingest_date = '2026-08-15'; ``` Delete-then-insert is a common, easy-to-reason-about way to make a date-partitioned load idempotent, but it is one option among several, not the universal answer. Depending on the destination and the workload, an `INSERT ... ON CONFLICT` upsert (or a `MERGE` statement), a full partition-replace operation, or a transactional staging-table-then-swap pattern may be a better fit - the property you are always aiming for is the same regardless of which mechanism you pick: rerunning the task for the same date leaves the table in the same end state every time. > 📌 **Remember:** Idempotency is not optional polish - it is what makes retries, backfills, and manual reruns safe instead of dangerous. Every task you write for Airflow should be designed so that running it twice for the same date is harmless. ### Alerting ```python def notify_failure(context): task_id = context["task_instance"].task_id execution_date = context["ds"] # send to Slack, PagerDuty, or email here print(f"ALERT: {task_id} failed for {execution_date}") default_args = { "on_failure_callback": notify_failure, "retries": 3, } ``` `on_failure_callback` fires after all retries are exhausted, so it signals a genuine problem rather than a transient blip that already resolved itself. > 💡 **Practice:** Modify the `notify_failure` function above to also log which upstream task, if any, the failed task depended on. This is the first piece of information an on-call engineer needs when debugging at 3 AM. **Concept check:** A task appends rows without checking for existing data for that date, and Airflow retries it after a partial failure. What goes wrong, and how would you rewrite the task to prevent it? ---
It is 3 AM at a company like Swiggy. A cron job that pulls yesterday's orders from Postgres into the warehouse silently ...
Before writing a single DAG, it helps to know what is actually running when you use Airflow, because most beginner confu...
Every DAG file is a Python script that Airflow's Scheduler reads and parses. Getting the DAG definition right - the argu...
Tasks in a DAG frequently need to share small values - a row count, a file path, a boolean flag - with each other. Airfl...
Not every dependency lives inside a single DAG. Sometimes a pipeline needs to wait for a file that another team drops in...
A pipeline that runs correctly once is easy. A pipeline that runs correctly every single day, including the days somethi...
Backfilling means running a DAG for a range of past dates - useful when you deploy a new pipeline and want three months ...
Everything so far has been about orchestrating a pipeline. This section is about the actual ingestion strategy the tasks...
Everything up to this point ensures a pipeline runs reliably. None of it guarantees the data it moves is actually correc...
Airflow can run many tasks in parallel, across many DAGs, without any extra configuration - and that is exactly the prob...
Alerting (covered earlier) tells you when something has already broken. Observability is the broader practice of being a...
Airflow debugging is a skill in its own right, separate from knowing the concepts. The four scenarios below cover the mo...
Prerequisites: Docker and Docker Compose installed, basic familiarity with Python and SQL from the earlier modules in th...
Concept What it does Common syntax / setting DAG definition Declares a pipeline's schedule and tasks scheduleinterval / ...
Deploying a new DAG without explicitly setting catchup=False can cause Airflow to trigger a historical run for every day...
Aligns directly with DevOps, Site Reliability (SRE), and Platform Engineering job descriptions.