This is where your pipeline starts looking professional: Airflow orchestrates when and in what order things run, dbt owns how data is transformed once it's in the warehouse.
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
with DAG(
dag_id="orders_pipeline",
start_date=datetime(2026, 1, 1),
schedule="0 2 * * *", # every day at 2am
catchup=False,
) as dag:
extract = PythonOperator(task_id="extract", python_callable=extract_orders)
transform = PythonOperator(task_id="transform", python_callable=transform_orders)
load = PythonOperator(task_id="load", python_callable=load_orders)
extract >> transform >> load # define execution orderschedule is a cron expression (or preset like "@daily"); catchup=False stops Airflow from immediately running every missed interval since start_date.The "A" in DAG stands for "acyclic," meaning no loops. If task B depended on task A, and task A also depended on task B, they'd wait on each other forever and nothing would ever run. Airflow checks for this when it loads your DAG, and simply refuses to run one that contains a loop — which is why the >> arrows can only point forward.
The TaskFlow API — the same DAG above, written with the newer decorator-based style that's largely replaced raw PythonOperator in current Airflow codebases:
from airflow.decorators import dag, task
from datetime import datetime
@dag(dag_id="orders_pipeline", start_date=datetime(2026, 1, 1), schedule="0 2 * * *", catchup=False)
def orders_pipeline():
@task
def extract():
return fetch_orders_from_api()
@task
def transform(raw_data):
return clean(raw_data)
@task
def load(clean_data):
write_to_warehouse(clean_data)
load(transform(extract())) # dependencies are inferred from the function calls themselves
orders_pipeline()Notice there's no >> anywhere in this version. Calling transform(extract()) both passes the data along and sets up the dependency, in one line. Return values get wired through XCom (covered below) automatically, so you don't need to push and pull them by hand.
PythonOperator, BashOperator, PostgresOperator, S3ToRedshiftOperator...). Task — an operator instantiated inside a specific DAG with a unique task_id.A sensor in practice — waiting for an upstream file to actually exist in S3 before starting the extract, instead of guessing a fixed delay:
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
wait_for_file = S3KeySensor(
task_id="wait_for_orders_export",
bucket_name="my-de-bucket",
bucket_key="raw/orders/dt={{ ds }}/orders.json", # {{ ds }} is the run's logical date (below)
poke_interval=60, # check every 60 seconds
timeout=60 * 60, # give up after 1 hour and fail the task
mode="reschedule", # free up the worker slot between checks instead of blocking it
)
wait_for_file >> extract >> transform >> loadmode="reschedule" matters once you have a lot of DAGs running: the default, mode="poke", holds onto a worker slot for the entire wait — which starves other tasks if you have many sensors waiting at once. reschedule gives up the slot between checks so other tasks can use it. And {{ ds }} is a template — Airflow fills it in with the DAG run's actual date right before the task runs, which is how the exact same DAG definition ends up looking at a different S3 file every single day.
extract >> transform >> load # linear chain
extract >> [validate, profile] >> load # fan-out then fan-in
task_a.set_downstream(task_b) # equivalent to task_a >> task_b>>/<<) define the DAG's edges — a task only starts once all its upstream dependencies succeed.Fan-out/fan-in, worked example — extracting from three independent sources in parallel, then only proceeding once all three are done:
extract_orders = PythonOperator(task_id="extract_orders", python_callable=extract_orders_fn)
extract_customers = PythonOperator(task_id="extract_customers", python_callable=extract_customers_fn)
extract_products = PythonOperator(task_id="extract_products", python_callable=extract_products_fn)
merge_and_load = PythonOperator(task_id="merge_and_load", python_callable=merge_and_load_fn)
[extract_orders, extract_customers, extract_products] >> merge_and_load
# all three extracts run concurrently (up to your worker/parallelism limits);
# merge_and_load only starts once every one of them has succeededBy default, if any of the three fails, merge_and_load gets marked upstream_failed and never runs at all — Airflow's default rule is "every upstream task must succeed first." You can loosen this with trigger_rule="all_done" (run no matter what happened upstream) or "one_success" (run once at least one upstream task succeeded), for cases where that strict default doesn't fit — like a cleanup task that should run whether or not the rest of the pipeline succeeded.
The Scheduler decides what should run next. The Executor decides how and where it actually runs — one task at a time, in parallel on one machine, or spread across a whole cluster.
airflow.cfg or an environment variable), not something you touch inside a DAG file.AIRFLOW__CORE__EXECUTOR=<name>, and for the distributed ones, point Airflow at the extra infrastructure they need (a real database, a message broker, or a Kubernetes cluster).| Executor | Parallelism | Extra infra needed | Good for |
|---|---|---|---|
| SequentialExecutor | None — one task at a time | None (works with SQLite) | The out-of-the-box default; local testing only |
| LocalExecutor | Multiple processes, one machine | A real database (Postgres/MySQL, not SQLite) | Small-to-medium deployments — exactly your Week 6 docker-compose stack |
| CeleryExecutor | Distributed across worker machines | Celery + a broker (Redis/RabbitMQ) + a real database | Scaling beyond one machine, with a fairly steady workload |
| KubernetesExecutor | One pod per task, elastic | A Kubernetes cluster | Bursty/variable workloads, per-task resource isolation |
| CeleryKubernetesExecutor | Hybrid — routes per task | Both Celery and Kubernetes | Mixed workloads where only some tasks need pod isolation |
You've actually already used one of these, without giving it a name: once your Week 6 docker-compose stack is wired to Postgres instead of the SQLite default, it's running on LocalExecutor — multiple tasks running at the same time, as separate processes, all on that one container.
# SequentialExecutor -> LocalExecutor: swap the executor and point at a real database
# (in your docker-compose.yml environment block from Week 6)
AIRFLOW__CORE__EXECUTOR: LocalExecutor
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://postgres:postgres@postgres:5432/airflowScaling past one machine with CeleryExecutor: add a message broker, and Airflow starts handing tasks out to a pool of separate worker processes — possibly on separate machines entirely — instead of running everything on the scheduler's own machine:
AIRFLOW__CORE__EXECUTOR: CeleryExecutor
AIRFLOW__CELERY__BROKER_URL: redis://redis:6379/0
AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://postgres:postgres@postgres:5432/airflow
# then run one or more worker containers alongside the scheduler:
# airflow celery workerPer-task resource control with KubernetesExecutor: this is what you'd reach for when different tasks need very different amounts of CPU or memory, since each task gets its own freshly-started pod instead of sharing one fixed-size worker:
train_model = PythonOperator(
task_id="train_model",
python_callable=train_model_fn,
executor_config={
"KubernetesExecutor": {
"request_memory": "4Gi",
"request_cpu": "2",
}
},
)
# a lightweight task elsewhere in the same DAG can stay on default, much smaller resources —
# each task's pod is sized independently instead of every task competing for one shared workerThe trade-off to remember: LocalExecutor is the simplest, but it's limited to what one machine can handle. CeleryExecutor scales well, but you're paying for a fleet of workers even when they're sitting idle. KubernetesExecutor scales up and down automatically (a pod only exists while its task is running), but each task pays a small startup cost, and you need a Kubernetes cluster to begin with. Most teams start on LocalExecutor and only move to Celery or Kubernetes once they've genuinely outgrown it — don't over-think this choice on day one.
extract = PythonOperator(
task_id="extract",
python_callable=extract_orders,
retries=3,
retry_delay=timedelta(minutes=5),
retry_exponential_backoff=True,
)Here's what this config actually looks like in practice: say attempt 1 fails at 2:00am. With retry_delay=5min and retry_exponential_backoff=True, attempt 2 fires around 2:05am, attempt 3 around 2:15am (the delay roughly doubles each time), and attempt 4 around 2:35am. If attempt 4 also fails, the task is marked failed for good. Airflow doesn't retry forever.
Retries can also be set at the DAG level as a default every task inherits, overridden per-task where needed:
default_args = {
"retries": 2,
"retry_delay": timedelta(minutes=5),
}
with DAG(dag_id="orders_pipeline", default_args=default_args, ...) as dag:
extract = PythonOperator(task_id="extract", python_callable=extract_orders, retries=5) # overrides the default
load = PythonOperator(task_id="load", python_callable=load_orders) # uses the default: retries=2A good rule of thumb: give tasks that depend on flaky networks (like API calls) more retries. Give tasks where a retry genuinely can't help — like one that failed because the SQL itself has a bug — retries=0 instead. Retrying a real bug just delays the alert that tells you it's broken.
def extract(**context):
row_count = run_extraction()
context["ti"].xcom_push(key="row_count", value=row_count)
def notify(**context):
count = context["ti"].xcom_pull(task_ids="extract", key="row_count")
print(f"Extracted {count} rows")@task TaskFlow API, return values are auto-pushed to XCom and function arguments auto-pull them — less boilerplate than manual push/pull.Where does an XCom value actually live? Every value you push gets saved as a row in Airflow's own metadata database, tagged with the DAG, task, and run it belongs to. That's exactly why "don't put large data in XCom" isn't just a style preference — pushing a 500MB DataFrame would try to write 500MB into Airflow's own database, which is sized for small operational data, not big payloads. It can genuinely slow down or crash the scheduler.
# WRONG: pushes the entire dataset through XCom
@task
def extract():
df = fetch_all_orders() # could be millions of rows
return df # Airflow tries to serialize/store the whole thing
# RIGHT: do the heavy lifting outside Airflow's metadata store, pass only a pointer through XCom
@task
def extract():
df = fetch_all_orders()
path = f"s3://my-de-bucket/raw/orders/{ds}.parquet"
df.to_parquet(path)
return path # XCom carries a small string, not the data itself
@task
def transform(s3_path):
df = pd.read_parquet(s3_path) # the next task re-reads from S3 using the pointer
...airflow dags backfill orders_pipeline \
--start-date 2026-01-01 \
--end-date 2026-01-31Logical date, concretely — this is the single most confusing Airflow concept for newcomers, so it's worth a direct example. A daily DAG scheduled @daily that's meant to process "yesterday's orders":
# the run that Airflow labels logical_date = 2026-08-17
# does NOT execute on Aug 17 — it executes at the START of Aug 18
# (Airflow waits until a full day has elapsed before running the interval that covers it)
@task
def extract(**context):
logical_date = context["ds"] # "2026-08-17" — the date this run represents
url = f"https://api.example.com/orders?date={logical_date}"
return fetch(url)So when you backfill with --start-date 2026-01-01 --end-date 2026-01-31, Airflow runs 31 separate DAG runs — one for each day from Jan 1 through Jan 31 — and each one pulls and loads that specific day's data. It's not 31 runs that all just grab "today's" data. That's what makes backfill genuinely useful: fix a bug in your extraction code, then backfill, and every affected day gets correctly reprocessed with the fixed code and its own correct date.
on_failure_callback on a DAG/task to notify Slack/email/PagerDuty when something breaks.def notify_slack_on_failure(context):
task_id = context["task_instance"].task_id
dag_id = context["dag"].dag_id
log_url = context["task_instance"].log_url
send_slack_message(f"❌ {dag_id}.{task_id} failed. Logs: {log_url}")
with DAG(
dag_id="orders_pipeline",
default_args={"on_failure_callback": notify_slack_on_failure},
...
) as dag:
...on_failure_callback gets handed a bundle of info about the failed run — which task, which DAG, a direct link to its logs — so the alert can point a human straight at the problem, instead of just saying "something broke." Set it as a default_arg like above, and every task in the DAG picks it up automatically.
Reading the Grid view day to day is simple once you know the colors: green means success, red means failure, orange/yellow means up_for_retry, and light-red/pink means upstream_failed — the task itself never even ran, because something it depends on failed first. That last one trips people up the most: don't waste time debugging the pink task — go find the actual red one further upstream instead.
In Airflow, the last step is typically a BashOperator (or DbtRunOperator from the Cosmos/astronomer-cosmos package) that runs dbt run and dbt test as a task in the same DAG that just loaded raw data.
from airflow.operators.bash import BashOperator
run_dbt = BashOperator(
task_id="dbt_run",
bash_command="cd /opt/airflow/dbt_project && dbt run",
)
test_dbt = BashOperator(
task_id="dbt_test",
bash_command="cd /opt/airflow/dbt_project && dbt test",
)
extract >> validate >> transform >> load >> run_dbt >> test_dbtNotice dbt test is its own task, running after dbt run, instead of being bundled into it. That split matters — a failed test should look visibly different in the Airflow UI from a failed build, and you might want to handle each differently (alert immediately on a broken build, but only warn on some test failures). This six-task chain — extract, validate, transform, load, dbt run, dbt test — is basically the DAG you're building for the Week 12 capstone, just with real connections instead of placeholders.
# models/staging/sources.yml
sources:
- name: raw
schema: raw
tables:
- name: orders
- name: customersSources are just a list of the raw tables dbt is allowed to read from — tables that your Airflow extract-load step has already filled in. They let you write {{ source('raw', 'orders') }} in a model instead of hardcoding a table name, and give you a place to add freshness checks.
# extending sources.yml with a freshness check
sources:
- name: raw
schema: raw
tables:
- name: orders
loaded_at_field: _loaded_at
freshness:
warn_after: {count: 12, period: hour}
error_after: {count: 24, period: hour}Running dbt source freshness checks the newest _loaded_at timestamp in the raw table, and warns or errors if it's too old. This catches the case where Airflow's extract task quietly stopped working — say, a credential expired — without anything crashing loudly enough to alert anyone. It's often the very first sign that something upstream is broken, before any model built on top shows visibly wrong numbers.
There's another reason to always use {{ source(...) }} instead of a hardcoded table name: it's what lets the raw layer show up as the starting point of the lineage graph (below). dbt can only draw the full picture, from raw source all the way to final mart, if every single hop is declared through source()/ref().
-- models/staging/stg_orders.sql
SELECT
order_id,
customer_id,
CAST(order_date AS date) AS order_date,
total_amount::numeric AS total_amount
FROM {{ source('raw', 'orders') }}
WHERE total_amount >= 0
-- models/marts/fct_orders.sql
SELECT
o.order_id,
o.customer_id,
o.order_date,
o.total_amount
FROM {{ ref('stg_orders') }} oSELECT statement in a .sql file — dbt compiles and runs it, materializing it as a table or view.{{ ref('model_name') }} links models together and lets dbt build them in the correct dependency order automatically.staging/ (cleaned, 1:1 with source) → marts/ (fact/dim, business logic applied).Materializations — a config choice controlling exactly what dbt turns a model into when it runs, set per-model or as a project default:
-- models/staging/stg_orders.sql
{{ config(materialized='view') }} -- cheap staging models: just a view, always fresh
SELECT order_id, customer_id, CAST(order_date AS date) AS order_date
FROM {{ source('raw', 'orders') }}
-- models/marts/fct_orders.sql
{{ config(materialized='table') }} -- expensive-to-compute marts: a real table, faster to query
SELECT ...
FROM {{ ref('stg_orders') }}
-- models/marts/fct_orders_incremental.sql
{{ config(materialized='incremental', unique_key='order_id') }}
SELECT * FROM {{ ref('stg_orders') }}
{% if is_incremental() %}
WHERE order_date > (SELECT MAX(order_date) FROM {{ this }}) -- only process new rows
{% endif %}view costs nothing to build, but re-runs its query every single time someone reads it. table is precomputed and fast to query, but needs a full rebuild on every dbt run. incremental is the middle ground: it's a real table, but after the first full build, later runs only process new or changed rows, using the is_incremental() block — that's the watermark pattern from Week 4, built into dbt. Switch to incremental once a mart's full rebuild starts taking minutes instead of seconds.
# models/staging/schema.yml
models:
- name: stg_orders
columns:
- name: order_id
tests: [unique, not_null]
- name: customer_id
tests:
- not_null
- relationships:
to: ref('stg_customers')
field: customer_idRun these with dbt test. This is really just a formal version of the "data validation" idea from Week 4. unique, not_null, accepted_values, and relationships all come built in; write a custom SQL test for anything more specific.
What's actually happening under the hood: tests: [unique, not_null] gets turned into real SQL that dbt runs, checking that it returns zero rows. A custom test is just that same idea, written by hand:
-- tests/assert_positive_order_amounts.sql
-- a "singular" test: any row this query returns counts as a FAILURE
SELECT order_id, total_amount
FROM {{ ref('stg_orders') }}
WHERE total_amount < 0-- accepted_values, another built-in generic test
models:
- name: stg_orders
columns:
- name: order_status
tests:
- accepted_values:
values: ['pending', 'completed', 'cancelled', 'refunded']Run dbt test as part of the same Airflow task that runs dbt run (or right after it) — that's what turns "the pipeline ran" into "the pipeline ran and the output is trustworthy," which is the real answer to "how do you know your data is correct."
-- macros/cents_to_dollars.sql
{% macro cents_to_dollars(column_name) %}
({{ column_name }} / 100.0)
{% endmacro %}
-- usage in a model
SELECT order_id, {{ cents_to_dollars('amount_cents') }} AS amount_dollars
FROM {{ source('raw', 'orders') }}Macros are reusable SQL snippets written with Jinja templating — the dbt equivalent of a function, so you're not copy-pasting the same transformation logic across models.
Here's a more realistic macro — most dbt projects have something like this, to keep order-tier bucketing consistent across models. You'll recognize the logic from Week 1-2's CASE expression section; it's the exact same idea, just turned into something reusable:
-- macros/order_tier.sql
{% macro order_tier(amount_column) %}
CASE
WHEN {{ amount_column }} >= 500 THEN 'high'
WHEN {{ amount_column }} >= 100 THEN 'medium'
ELSE 'low'
END
{% endmacro %}
-- used identically in two different models, guaranteed to stay in sync
-- models/marts/fct_orders.sql
SELECT order_id, total_amount, {{ order_tier('total_amount') }} AS tier FROM {{ ref('stg_orders') }}
-- models/marts/fct_refunds.sql
SELECT refund_id, refund_amount, {{ order_tier('refund_amount') }} AS tier FROM {{ ref('stg_refunds') }}The real value here isn't just saving typing — it's that when the business changes the "high" threshold from 500 to 750, there's exactly one file to edit. Every model using the macro picks up the change on its next dbt run, instead of you having to hunt down and update every copy-pasted CASE statement by hand.
-- snapshots/customers_snapshot.sql
{% snapshot customers_snapshot %}
{{
config(
target_schema='snapshots',
unique_key='customer_id',
strategy='timestamp',
updated_at='updated_at',
)
}}
SELECT * FROM {{ source('raw', 'customers') }}
{% endsnapshot %}Snapshots implement SCD Type 2 (Week 4) for you automatically. On every run, dbt compares the current source data to the snapshot table, and whenever a tracked row has changed, it inserts a new row with dbt_valid_from/dbt_valid_to — keeping the full history intact.
Here's what the resulting table looks like after customer 101's city changes from Delhi to Chicago — the same output as the hand-written SQL in Week 4's SCD section, just generated automatically by dbt snapshot on every scheduled run:
| dbt_scd_id | customer_id | city | dbt_valid_from | dbt_valid_to |
|---|---|---|---|---|
| abc123 | 101 | Delhi | 2024-01-01 | 2026-03-15 |
| def456 | 101 | Chicago | 2026-03-15 | NULL |
dbt has two ways to detect a change. strategy='timestamp' (shown above) compares an updated_at column you point it at. strategy='check' instead compares specific columns directly — like check_cols=['city', 'email'] — for sources that don't reliably keep an updated-at timestamp. Downstream models then filter with WHERE dbt_valid_to IS NULL to get only the current version of each row, exactly like is_current = true in the hand-rolled version.
dbt docs generate
dbt docs servedbt docs generate builds a searchable site documenting every model, column, test, and description you've written in schema.yml files.ref()/source() calls.Writing the descriptions that power the docs site — a well-documented schema.yml turns into a browsable data dictionary, not just a place to declare tests:
models:
- name: fct_orders
description: "One row per completed order. Grain: order_id. Excludes cancelled orders."
columns:
- name: order_id
description: "Primary key, matches the source system's order ID."
tests: [unique, not_null]
- name: net_revenue
description: "total_amount minus the 5% platform fee, in USD."The lineage graph can trace backward from any model or column to every upstream source it depends on, and forward to every downstream model that uses it. That makes it the fastest way to answer "if I change this column, what breaks?" before you make the edit — the pipeline's dependency structure is written down and visible, not something only one person happens to remember.