DAGs, dbt,
and integrations

What we're building today

• Schedule-aware dependencies with trigger_rule and catchup.
• Scalable runtime fan-out with dynamic task mapping.
• Safe task communication through XCom.
• Airflow and dbt integration at several levels.
• Production-safe connections, configuration, and testing.

Which upstream outcome starts a task?

• all_success is the default: every upstream task must succeed.
• all_failed is useful for alerts that run only after failure.
• one_success and one_failed react to any matching branch.
• none_failed allows successful and skipped upstream tasks.
• all_done is useful for unconditional cleanup and final notifications.

How states propagate

Quick reference

A run processes a completed period

• The DagRun for January 1 starts at the beginning of January 2.
• Airflow waits until the whole interval is complete.
• logical_date labels the data period, not the wall-clock start time.
• This behavior is intentional for batch processing.

When does the DagRun execute?

The default changed in Airflow 3

• Use explicit catchup=True when historical processing is intended.
• Use backfill when you need a controlled date range.

Anti-pattern: module-level loops

• The DAG Processor executes every DAG file in full during each scan.
• A top-level loop runs again on every parse.
• More entities mean more DAG objects and linearly increasing parse time.
• The code works, but it adds shared scheduler overhead.

for table in TABLES:  # Runs during every parse.
    dag_id = f"etl_{table}"
    dag = DAG(dag_id=dag_id, ...)
    globals()[dag_id] = dag

Parsing time is the key difference

External configuration is a compromise

• A factory can read YAML, JSON, or database configuration.
• Configuration is separated from code and easier to review.
• Reading still happens while the DAG file is parsed.
• Use it when the entity list changes infrequently.

Inspect values before reading logs

• Open Task Instance Details → XCom.
• Verify the source task_id, key, value, and data type.
• A missing key returns None; it does not raise an exception.
• For medium-sized values, use a custom backend such as S3.

[core]
xcom_backend = my_project.xcom_backends.S3XComBackend

Push and pull two independent keys

• Push raw_value = 42.
• Push source = "api".
• Pull each key separately in the downstream task.
• Print: Value 42 was received from api.

dbt needs orchestration context

• dbt excels at transformation but does not orchestrate extraction and loading.
• Airflow starts dbt after data is actually ready, not after a fixed delay.
• One DAG provides source-to-consumer observability.
• Retries can target the failed step instead of the whole pipeline.

What orchestration adds

Granularity changes the debugging experience

Never refresh BI before tests pass

• Render dbt tests as part of the Airflow graph.
• Make BI refresh depend on successful models and tests.
• Stop consumers when data quality fails.
• Notify with the exact failed model and Airflow log URL.

dbt_transform >> refresh_bi_dashboard() >> notify_slack()

Evolve from cron to orchestration

1. Start with one scheduled BashOperator.
2. Add a deferrable SqlSensor for source readiness.
3. Replace the shell task with a Cosmos task group.
4. Add a deliberately failing model and compare Grid views.
5. Send an alert containing task ID and log URL.

Centralize access to external systems

• A Connection stores host, login, password, port, and extras.
• DAG code references a stable conn_id.
• Credentials stay out of source code.
• Create connections in UI, CLI, environment variables, or a secrets backend.

hook = PostgresHook(postgres_conn_id="my_postgres")
count = hook.get_first("SELECT COUNT(*) FROM orders")

Create a Postgres connection

Runtime configuration, not credentials

• Connections answer “where and how do I connect?”
• Variables hold feature flags, thresholds, and target paths.
• Read Variables inside task bodies, not at module level.
• Use deserialize_json=True for structured configuration.

target_bucket = Variable.get("target_bucket")

Make HTTP failures explicit

@task
def fetch_api_data():
    response = requests.get("https://api.example.com/data")
    response.raise_for_status()  # 404/500 → HTTPError
    return response.json()

• requests.get() returns a Response even when the server answers with 404 or 500.
• raise_for_status() converts that status into an HTTPError.
• Airflow marks the task failed only when the exception escapes the task function.

Return values become XCom

@task
def fetch_api_data():
    ...
    return response.json()  # Stored as key="return_value"

@task
def process_data(data):
    print(data)

process_data(fetch_api_data())  # Equivalent to an xcom_pull

• TaskFlow serializes the returned value and writes it to the metadata database.
• The value must be serializable and is visible in Task Instance Details → XCom.
• A downstream argument reads the persistent XCom record; this is not an in-memory function call.

DataFrames do not belong in metadata DB

return dataframe

dataframe.to_parquet(path)
return path

Production checklist

• Keep parse-time code deterministic and lightweight.
• Use Connections and a secrets backend for credentials.
• Pass references through XCom, not large datasets.
• Prefer deferrable waiting and runtime task mapping.
• Test DAG imports and failure paths in CI.