ETL Pipelines with Apache Airflow and dbt – Complete Practical Guide

ETL (Extract, Transform, Load) pipelines are the circulatory system of a data organisation — moving data from source systems into analytics-ready storage, transforming it into the right shape and quality, and doing so reliably, automatically, and at scale. Every machine learning model is only as good as the data it is trained on, and every dashboard is only as trustworthy as the pipeline feeding it. This guide covers the practical implementation of ETL/ELT pipelines using Apache Airflow for orchestration and dbt for transformation — the two most widely used tools in modern data engineering.

This guide is the practical implementation companion to our Data Engineering Fundamentals guide (which covers the conceptual ETL vs ELT distinction and the modern data stack). The SQL transformation patterns used in dbt connect to our Advanced SQL guide. The pipelines built here feed the feature stores and ML workflows covered in our MLOps Interview Q&A. Python data manipulation patterns within pipeline tasks use the skills in our Pandas and NumPy Mastery guide.

Apache Airflow — Architecture and Core Concepts

Airflow is a platform for authoring, scheduling, and monitoring data pipelines as Directed Acyclic Graphs (DAGs). Its architecture has four components: the Scheduler (decides what runs when, submits task instances), the Executor (runs tasks — LocalExecutor for single-node, CeleryExecutor or KubernetesExecutor for distributed), the Metadata Database (stores DAG definitions, task states, logs — typically PostgreSQL), and the Web Server (the Airflow UI for monitoring and triggering runs).

Key Airflow concepts: A DAG is a Python file that defines tasks and their dependencies. A DAG Run is one execution of a DAG, identified by its execution_date (the logical date, not the actual run time). Task Instances are individual executions of a task within a DAG Run, with states: queued, running, success, failed, skipped, upstream_failed. XComs (cross-communications) let tasks pass small data (IDs, file paths, counts) to downstream tasks via the metadata database — not for large data, which should go through external storage (S3, GCS).

from airflow import DAG
from airflow.operators.python import PythonOperator, BranchPythonOperator
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator
from airflow.sensors.filesystem import FileSensor
from airflow.utils.trigger_rule import TriggerRule
from datetime import datetime, timedelta
import pandas as pd, requests, boto3, logging

logger = logging.getLogger(__name__)

default_args = {
    'owner':             'data_engineering',
    'depends_on_past':   False,
    'start_date':        datetime(2026, 9, 1),
    'retries':           3,
    'retry_delay':       timedelta(minutes=10),
    'retry_exponential_backoff': True,
    'max_retry_delay':   timedelta(hours=1),
    'email_on_failure':  True,
    'email_on_retry':    False,
    'email':             ['data-alerts@company.com'],
}

with DAG(
    dag_id='ecommerce_daily_pipeline',
    default_args=default_args,
    description='Daily sales ingestion, validation, and warehouse load',
    schedule_interval='30 5 * * *',   # 05:30 UTC daily (after source closes)
    catchup=False,
    max_active_runs=1,                 # prevent concurrent runs
    tags=['ecommerce', 'daily', 'critical'],
) as dag:

    def extract_orders(**context):
        ds       = context['ds']        # execution date: YYYY-MM-DD
        ds_nodash = context['ds_nodash'] # YYYYMMDD
        logger.info('Extracting orders for date: ' + ds)

        resp = requests.get(
            'https://internal-api.company.com/orders',
            params={'date': ds, 'status': 'all'},
            headers={'X-API-Key': '{{ var.value.orders_api_key }}'},
            timeout=60
        )
        resp.raise_for_status()
        data = resp.json()

        # Write to S3 raw zone
        s3 = boto3.client('s3')
        s3.put_object(
            Bucket='data-lake',
            Key='raw/orders/' + ds + '/orders.json',
            Body=resp.text,
            ContentType='application/json'
        )
        # Push row count to XCom for downstream validation
        context['ti'].xcom_push('row_count', len(data))
        context['ti'].xcom_push('s3_key', 'raw/orders/' + ds + '/orders.json')
        logger.info('Extracted ' + str(len(data)) + ' orders')
        return len(data)

    def validate_data(**context):
        ti        = context['ti']
        row_count = ti.xcom_pull(task_ids='extract_orders', key='row_count')
        ds        = context['ds']

        # Data quality checks
        issues = []
        if row_count == 0:
            issues.append('Zero rows extracted — source may be down')
        if row_count < 1000:
            issues.append('Row count ' + str(row_count) + ' below expected minimum of 1000')

        # Compare with yesterday (stored in Variable or DB)
        from airflow.models import Variable
        prev_count = int(Variable.get('orders_prev_count', default_var=0))
        if prev_count > 0 and abs(row_count - prev_count) / prev_count > 0.30:
            issues.append('Row count deviated >30% from yesterday (' +
                          str(prev_count) + ' vs ' + str(row_count) + ')')

        if issues:
            logger.error('Validation failed: ' + str(issues))
            raise ValueError('Data validation failed: ' + '; '.join(issues))

        Variable.set('orders_prev_count', row_count)
        logger.info('Validation passed: ' + str(row_count) + ' rows')
        return 'load_to_warehouse'   # for BranchPythonOperator pattern

    extract  = PythonOperator(task_id='extract_orders',  python_callable=extract_orders)
    validate = PythonOperator(task_id='validate_data',   python_callable=validate_data)

    load_staging = BashOperator(
        task_id='load_to_staging',
        bash_command=(
            'snowsql -a myaccount -u svc_airflow '
            '--query "COPY INTO raw.orders_staging '
            "FROM @s3_stage/raw/orders/{{ ds }}/ "
            "FILE_FORMAT=(TYPE=JSON) PURGE=FALSE;""
        )
    )

    run_dbt = BashOperator(
        task_id='run_dbt_transformations',
        bash_command=(
            'cd /opt/dbt/ecommerce && '
            'dbt run --select +marts.fct_daily_orders --vars '
            '\'{"run_date": "{{ ds }}"}\'  '
            '--profiles-dir /opt/dbt/profiles'
        )
    )

    test_dbt = BashOperator(
        task_id='test_dbt_models',
        bash_command=(
            'cd /opt/dbt/ecommerce && '
            'dbt test --select +marts.fct_daily_orders'
        )
    )

    notify_success = BashOperator(
        task_id='notify_success',
        bash_command='curl -X POST $SLACK_WEBHOOK -d '{"text":"Daily pipeline completed successfully for {{ ds }}"}' ',
        trigger_rule=TriggerRule.ALL_SUCCESS
    )

    notify_failure = BashOperator(
        task_id='notify_failure',
        bash_command='curl -X POST $SLACK_WEBHOOK -d '{"text":"ALERT: Daily pipeline FAILED for {{ ds }}"}' ',
        trigger_rule=TriggerRule.ONE_FAILED
    )

    extract >> validate >> load_staging >> run_dbt >> test_dbt
    test_dbt >> [notify_success, notify_failure]

dbt — Data Transformation Best Practices

dbt structures transformations into three layers: Staging (one-to-one with source tables — rename, cast, and light cleaning only), Intermediate (business logic joins and aggregations across staging models), and Marts (final business-ready tables consumed by BI tools and ML models — star schema fact and dimension tables). This layering makes each model’s purpose clear, keeps transformations composable, and enables incremental materialisation at the marts level.

-- models/staging/stg_orders.sql
-- Staging: clean + standardise raw source, nothing more

-- config(materialized='view', tags=['staging', 'daily'])

WITH source AS (
    SELECT * FROM raw.orders_staging
    WHERE _loaded_at >= DATEADD('day', -2, CURRENT_DATE)  -- safe reload window
),
renamed AS (
    SELECT
        order_id::VARCHAR(50)           AS order_id,
        customer_id::VARCHAR(50)        AS customer_id,
        CAST(created_at AS TIMESTAMP_NTZ) AS created_at,
        DATE(created_at)                AS order_date,
        UPPER(TRIM(status))             AS status,          -- normalise casing
        ROUND(subtotal::FLOAT, 2)       AS subtotal_usd,
        ROUND(tax_amount::FLOAT, 2)     AS tax_usd,
        ROUND(total_amount::FLOAT, 2)   AS total_usd,
        COALESCE(country_code, 'UNKNOWN') AS country_code,
        COALESCE(currency, 'USD')       AS currency,
        channel                         AS acquisition_channel,
        _loaded_at                      AS _loaded_at
    FROM source
    WHERE order_id IS NOT NULL          -- reject null PKs
      AND created_at IS NOT NULL
)
SELECT * FROM renamed;


-- models/marts/fct_daily_revenue.sql
-- Fact table: daily revenue aggregated by region and channel

-- config(
--   materialized='incremental',
--   unique_key=['revenue_date', 'country_code', 'channel'],
--   on_schema_change='sync_all_columns',
--   cluster_by=['revenue_date'],
--   tags=['marts', 'finance', 'daily']
-- )

WITH orders AS (
    SELECT * FROM stg_orders          -- ref('stg_orders') in real dbt
    WHERE status IN ('COMPLETED', 'DELIVERED')
    -- incremental filter: only process recent data on incremental runs
    -- is_incremental() check omitted here for readability
),
daily_agg AS (
    SELECT
        order_date                          AS revenue_date,
        country_code,
        acquisition_channel                 AS channel,
        COUNT(DISTINCT order_id)            AS n_orders,
        COUNT(DISTINCT customer_id)         AS n_customers,
        ROUND(SUM(subtotal_usd), 2)         AS gross_revenue_usd,
        ROUND(SUM(tax_usd), 2)              AS tax_collected_usd,
        ROUND(SUM(total_usd), 2)            AS net_revenue_usd,
        ROUND(AVG(total_usd), 2)            AS avg_order_value_usd,
        ROUND(SUM(total_usd) /
              NULLIF(COUNT(DISTINCT customer_id), 0), 2) AS revenue_per_customer
    FROM orders
    GROUP BY 1, 2, 3
)
SELECT
    {{ dbt_utils.generate_surrogate_key(['revenue_date', 'country_code', 'channel']) }}
        AS revenue_key,
    *,
    CURRENT_TIMESTAMP AS _dbt_updated_at
FROM daily_agg;

Data Quality Testing with dbt and Great Expectations

# schema.yml — dbt test definitions
# models:
#   - name: stg_orders
#     columns:
#       - name: order_id
#         tests: [unique, not_null]
#       - name: status
#         tests:
#           - accepted_values:
#               values: ['COMPLETED', 'PENDING', 'CANCELLED', 'DELIVERED', 'REFUNDED']
#       - name: total_usd
#         tests:
#           - not_null
#           - dbt_utils.expression_is_true:
#               expression: ">= 0"
#       - name: customer_id
#         tests:
#           - relationships:
#               to: ref('stg_customers')
#               field: customer_id

# Great Expectations for custom validation beyond dbt
import great_expectations as gx

context   = gx.get_context()
validator = context.sources.pandas_default.read_dataframe(df_orders)

validator.expect_column_values_to_not_be_null('order_id')
validator.expect_column_values_to_be_unique('order_id')
validator.expect_column_values_to_be_between('total_usd', min_value=0, max_value=500000)
validator.expect_column_values_to_be_in_set(
    'status', ['COMPLETED', 'PENDING', 'CANCELLED', 'DELIVERED', 'REFUNDED']
)
validator.expect_column_pair_values_A_to_be_greater_than_B(
    'total_usd', 'subtotal_usd', or_equal=True
)

results = validator.validate()
print('Validation passed:', results['success'])
if not results['success']:
    failed = [r for r in results['results'] if not r['success']]
    for r in failed:
        print('FAILED:', r['expectation_config']['expectation_type'])

For the full conceptual overview of ETL vs ELT, modern data stack architecture, and data lakehouse patterns, see our Data Engineering Fundamentals guide. For the SQL skills underlying dbt transformations — window functions, CTEs, optimisation — our Advanced SQL guide covers the techniques used in production dbt models. Data engineering interview questions on Airflow, Spark, Kafka, and data modelling are covered in our Data Engineering Interview Q&A. The ML pipelines that consume data produced by these ETL workflows are covered in our MLOps Interview Q&A.