Data Pipeline Patterns: ETL vs ELT, Streaming, Batch Processing, and Orchestration

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

// Debezium source connector — CDC from PostgreSQL
{
  "name": "postgres-cdc-source",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "secret",
    "database.dbname": "production",
    "database.server.name": "prod",
    "table.include.list": "public.orders,public.customers,public.products",
    "plugin.name": "pgoutput",
    "publication.autocreate.mode": "filtered",
    "decimal.handling.mode": "double",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false",
    "transforms.unwrap.delete.handling.mode": "rewrite"
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

// S3 sink connector — land raw events into the data lake
{
  "name": "s3-sink",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "tasks.max": "4",
    "topics": "prod.public.orders,prod.public.customers",
    "s3.region": "us-east-1",
    "s3.bucket.name": "data-lake-raw",
    "s3.part.size": "67108864",
    "storage.class": "io.confluent.connect.s3.storage.S3Storage",
    "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
    "flush.size": "100000",
    "rotate.interval.ms": "600000",
    "locale": "en_US",
    "timezone": "UTC",
    "timestamp.extractor": "RecordField",
    "timestamp.field": "updated_at",
    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "path.format": "'year'=YYYY/'month'=MM/'day'=dd/'hour'=HH",
    "locale": "en_US",
    "timezone": "UTC"
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

# PyFlink: real-time order aggregation
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, EnvironmentSettings

env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(4)

settings = EnvironmentSettings.new_instance().in_streaming_mode().build()
t_env = StreamTableEnvironment.create(env, environment_settings=settings)

# Create Kafka source table
t_env.execute_sql("""
    CREATE TABLE orders_raw (
        order_id        STRING,
        customer_id     STRING,
        product_id      STRING,
        amount          DECIMAL(10, 2),
        status          STRING,
        event_time      TIMESTAMP(3),
        WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'prod.public.orders',
        'properties.bootstrap.servers' = 'kafka:9092',
        'properties.group.id' = 'flink-orders-agg',
        'scan.startup.mode' = 'latest-offset',
        'format' = 'json',
        'json.timestamp-format.standard' = 'ISO-8601'
    )
""")

# Create output sink (e.g., write results back to Kafka or a database)
t_env.execute_sql("""
    CREATE TABLE order_aggregates (
        window_start    TIMESTAMP(3),
        window_end      TIMESTAMP(3),
        customer_id     STRING,
        order_count     BIGINT,
        total_amount    DECIMAL(10, 2),
        avg_amount      DECIMAL(10, 2),
        PRIMARY KEY (window_start, customer_id) NOT ENFORCED
    ) WITH (
        'connector' = 'upsert-kafka',
        'topic' = 'order-aggregates',
        'properties.bootstrap.servers' = 'kafka:9092',
        'key.format' = 'json',
        'value.format' = 'json'
    )
""")

# Tumbling window aggregation — 5-minute windows
t_env.execute_sql("""
    INSERT INTO order_aggregates
    SELECT
        TUMBLE_START(event_time, INTERVAL '5' MINUTE) AS window_start,
        TUMBLE_END(event_time,   INTERVAL '5' MINUTE) AS window_end,
        customer_id,
        COUNT(*)              AS order_count,
        SUM(amount)           AS total_amount,
        AVG(amount)           AS avg_amount
    FROM orders_raw
    WHERE status = 'COMPLETED'
    GROUP BY
        TUMBLE(event_time, INTERVAL '5' MINUTE),
        customer_id
""")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

# Flink watermark strategy with late data handling
t_env.execute_sql("""
    CREATE TABLE orders_with_late (
        order_id    STRING,
        amount      DECIMAL(10, 2),
        event_time  TIMESTAMP(3),
        -- Watermark: wait 30 seconds for late events
        WATERMARK FOR event_time AS event_time - INTERVAL '30' SECOND
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'orders',
        'properties.bootstrap.servers' = 'kafka:9092',
        'format' = 'json'
    )
""")

# Session windows — group events within 10 minutes of each other
t_env.execute_sql("""
    SELECT
        SESSION_START(event_time, INTERVAL '10' MINUTE) AS session_start,
        SESSION_END(event_time,   INTERVAL '10' MINUTE) AS session_end,
        customer_id,
        COUNT(*) AS orders_in_session,
        SUM(amount) AS session_value
    FROM orders_with_late
    GROUP BY
        SESSION(event_time, INTERVAL '10' MINUTE),
        customer_id
""")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

# Process Debezium CDC events — maintain a materialized view
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment

t_env = StreamTableEnvironment.create(...)

# Debezium produces: op="c" (create), "u" (update), "d" (delete), "r" (read/snapshot)
t_env.execute_sql("""
    CREATE TABLE customers_cdc (
        id          BIGINT,
        name        STRING,
        email       STRING,
        created_at  TIMESTAMP(3),
        op          STRING METADATA FROM 'value.op' VIRTUAL,
        PRIMARY KEY (id) NOT ENFORCED
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'prod.public.customers',
        'properties.bootstrap.servers' = 'kafka:9092',
        'format' = 'debezium-json'
    )
""")

# Upsert into a materialized table in a database
t_env.execute_sql("""
    CREATE TABLE customers_materialized (
        id      BIGINT PRIMARY KEY NOT ENFORCED,
        name    STRING,
        email   STRING
    ) WITH (
        'connector' = 'jdbc',
        'url' = 'jdbc:postgresql://warehouse:5432/analytics',
        'table-name' = 'customers'
    )
""")

t_env.execute_sql("""
    INSERT INTO customers_materialized
    SELECT id, name, email FROM customers_cdc
""")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

name: 'my_project'
version: '1.0.0'
config-version: 2

profile: 'my_project'

model-paths: ["models"]
test-paths: ["tests"]
snapshot-paths: ["snapshots"]
macro-paths: ["macros"]

models:
  my_project:
    staging:
      +materialized: view          # Staging models are views — no storage cost
      +schema: staging
    intermediate:
      +materialized: ephemeral     # Inlined as CTEs, not materialized
    marts:
      +materialized: table         # Final marts are tables for query performance
      +schema: marts

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

-- models/staging/stg_orders.sql
-- Rename columns, cast types, filter junk, nothing else

with source as (
    select * from {{ source('raw', 'orders') }}
),

renamed as (
    select
        -- Keys
        id::varchar                         as order_id,
        customer_id::varchar                as customer_id,
        product_id::varchar                 as product_id,

        -- Amounts
        amount::numeric(10,2)               as order_amount_usd,

        -- Dates (normalize to UTC timestamp)
        created_at at time zone 'UTC'       as created_at,
        updated_at at time zone 'UTC'       as updated_at,

        -- Status normalization
        lower(trim(status))                 as order_status,

        -- Metadata
        _loaded_at                          as _loaded_at

    from source
    where id is not null
      and amount > 0          -- Filter obviously invalid records
),

final as (
    select *,
        case
            when order_status in ('complete', 'completed', 'done') then 'completed'
            when order_status in ('cancel', 'cancelled', 'void')   then 'cancelled'
            else order_status
        end as order_status_normalized
    from renamed
)

select * from final

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

-- models/intermediate/int_orders_enriched.sql
-- Join orders with customer and product data

with orders as (
    select * from {{ ref('stg_orders') }}
),

customers as (
    select * from {{ ref('stg_customers') }}
),

products as (
    select * from {{ ref('stg_products') }}
),

enriched as (
    select
        o.order_id,
        o.order_amount_usd,
        o.order_status_normalized,
        o.created_at,

        -- Customer attributes
        c.customer_id,
        c.customer_name,
        c.customer_email,
        c.customer_segment,
        c.first_order_date,

        -- Product attributes
        p.product_id,
        p.product_name,
        p.product_category,
        p.product_cost_usd,

        -- Derived
        o.order_amount_usd - p.product_cost_usd   as gross_margin_usd,
        date_trunc('month', o.created_at)::date   as order_month

    from orders o
    left join customers c using (customer_id)
    left join products  p using (product_id)
)

select * from enriched

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

-- models/marts/customer_lifetime_value.sql
-- Final analytical table: one row per customer

{{
    config(
        materialized='table',
        sort='customer_segment',
        dist='customer_id'    -- Redshift distribution key
    )
}}

with orders as (
    select * from {{ ref('int_orders_enriched') }}
    where order_status_normalized = 'completed'
),

clv as (
    select
        customer_id,
        customer_name,
        customer_email,
        customer_segment,
        first_order_date,

        count(distinct order_id)                            as total_orders,
        sum(order_amount_usd)                               as lifetime_revenue_usd,
        sum(gross_margin_usd)                               as lifetime_margin_usd,
        avg(order_amount_usd)                               as avg_order_value_usd,
        max(created_at)                                     as last_order_date,
        max(created_at)::date - first_order_date::date      as customer_age_days,

        -- Cohort month
        date_trunc('month', first_order_date)::date         as acquisition_month,

        -- Recency (days since last order)
        current_date - max(created_at)::date                as days_since_last_order

    from orders
    group by 1, 2, 3, 4, 5
),

-- Add churn flag
final as (
    select
        *,
        case when days_since_last_order > 90 then true else false end as is_churned,
        lifetime_revenue_usd / nullif(customer_age_days, 0) * 365     as annualized_revenue_usd
    from clv
)

select * from final

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# models/staging/_sources.yml
version: 2

sources:
  - name: raw
    database: analytics
    schema: raw_data
    tables:
      - name: orders
        loaded_at_field: _loaded_at
        freshness:
          warn_after: {count: 6, period: hour}
          error_after: {count: 24, period: hour}
        columns:
          - name: id
            tests:
              - not_null
              - unique
          - name: amount
            tests:
              - not_null
              - dbt_utils.expression_is_true:
                  expression: ">= 0"
          - name: status
            tests:
              - accepted_values:
                  values: ['pending', 'processing', 'completed', 'cancelled']

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

# models/marts/_schema.yml
version: 2

models:
  - name: customer_lifetime_value
    description: "One row per customer with aggregated lifetime metrics"
    columns:
      - name: customer_id
        tests:
          - not_null
          - unique
      - name: lifetime_revenue_usd
        tests:
          - not_null
          - dbt_utils.expression_is_true:
              expression: ">= 0"
      - name: customer_segment
        tests:
          - not_null
          - accepted_values:
              values: ['enterprise', 'mid-market', 'smb', 'self-serve']

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

-- snapshots/orders_snapshot.sql
-- Track historical state of orders (Type 2 SCD)

{% snapshot orders_snapshot %}
{{
    config(
        target_schema='snapshots',
        unique_key='order_id',
        strategy='timestamp',
        updated_at='updated_at',
        invalidate_hard_deletes=True
    )
}}

select * from {{ source('raw', 'orders') }}

{% endsnapshot %}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# dbt commands
dbt run                          # Run all models
dbt run --select staging.*       # Run staging models only
dbt run --select +marts.customer_lifetime_value  # Run with all upstream deps
dbt test                         # Run all tests
dbt test --select stg_orders     # Test a specific model
dbt snapshot                     # Run snapshots
dbt source freshness             # Check source freshness
dbt docs generate && dbt docs serve   # Build and serve documentation
dbt compile                      # Compile SQL without running
dbt debug                        # Test connection and config

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106

# dags/daily_etl.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.transfers.s3_to_redshift import S3ToRedshiftOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.utils.task_group import TaskGroup

default_args = {
    'owner': 'data-engineering',
    'depends_on_past': False,
    'email_on_failure': True,
    'email': ['data-alerts@company.com'],
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
    'execution_timeout': timedelta(hours=2),
}

with DAG(
    dag_id='daily_data_pipeline',
    default_args=default_args,
    description='Extract from Postgres, load to S3, transform with dbt',
    schedule_interval='0 2 * * *',    # 2 AM daily
    start_date=datetime(2026, 1, 1),
    catchup=False,                     # Don't backfill missed runs
    max_active_runs=1,                 # Prevent concurrent runs
    tags=['production', 'daily'],
) as dag:

    # --- Extract phase ---
    with TaskGroup('extract', tooltip='Extract data from sources') as extract_group:

        extract_orders = BashOperator(
            task_id='extract_orders',
            bash_command="""
                psql $SOURCE_DB_CONN -c "\COPY (
                    SELECT * FROM orders
                    WHERE updated_at >= '{{ ds }}' AND updated_at < '{{ next_ds }}'
                ) TO STDOUT CSV HEADER" | \
                aws s3 cp - s3://data-lake/raw/orders/date={{ ds }}/orders.csv
            """,
            env={'SOURCE_DB_CONN': '{{ var.value.source_db_conn }}'},
        )

        extract_customers = BashOperator(
            task_id='extract_customers',
            bash_command="""
                psql $SOURCE_DB_CONN -c "\COPY customers TO STDOUT CSV HEADER" | \
                aws s3 cp - s3://data-lake/raw/customers/date={{ ds }}/customers.csv
            """,
            env={'SOURCE_DB_CONN': '{{ var.value.source_db_conn }}'},
        )

    # --- Load phase ---
    with TaskGroup('load', tooltip='Load raw data into warehouse') as load_group:

        load_orders = S3ToRedshiftOperator(
            task_id='load_orders_to_warehouse',
            schema='raw_data',
            table='orders',
            s3_bucket='data-lake',
            s3_key='raw/orders/date={{ ds }}/orders.csv',
            copy_options=[
                'CSV HEADER',
                'IGNOREHEADER 1',
                "TIMEFORMAT 'auto'",
                'ACCEPTINVCHARS',
            ],
            redshift_conn_id='redshift_default',
            aws_conn_id='aws_default',
        )

        load_customers = S3ToRedshiftOperator(
            task_id='load_customers_to_warehouse',
            schema='raw_data',
            table='customers',
            s3_bucket='data-lake',
            s3_key='raw/customers/date={{ ds }}/customers.csv',
            copy_options=['CSV HEADER', 'IGNOREHEADER 1'],
            redshift_conn_id='redshift_default',
            aws_conn_id='aws_default',
        )

    # --- Transform phase (dbt) ---
    with TaskGroup('transform', tooltip='Run dbt models') as transform_group:

        dbt_run_staging = BashOperator(
            task_id='dbt_staging',
            bash_command='cd /opt/dbt/my_project && dbt run --select staging.*',
        )

        dbt_run_marts = BashOperator(
            task_id='dbt_marts',
            bash_command='cd /opt/dbt/my_project && dbt run --select marts.*',
        )

        dbt_test = BashOperator(
            task_id='dbt_test',
            bash_command='cd /opt/dbt/my_project && dbt test',
        )

        dbt_run_staging >> dbt_run_marts >> dbt_test

    # --- Dependencies ---
    extract_group >> load_group >> transform_group

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

# Use XCom for small data passing between tasks
def check_row_count(**context):
    count = get_row_count_from_db()
    context['ti'].xcom_push(key='row_count', value=count)
    if count == 0:
        raise ValueError("No rows extracted — aborting pipeline")

def validate_counts(**context):
    count = context['ti'].xcom_pull(task_ids='check_count', key='row_count')
    print(f"Processing {count} rows")

# Use sensors for external dependencies
from airflow.sensors.s3_key_sensor import S3KeySensor

wait_for_upstream_file = S3KeySensor(
    task_id='wait_for_upstream',
    bucket_name='data-lake',
    bucket_key='upstream/ready/{{ ds }}/_SUCCESS',
    aws_conn_id='aws_default',
    timeout=3600,           # Wait up to 1 hour
    poke_interval=60,       # Check every minute
    mode='reschedule',      # Release worker slot while waiting
)

# Use branching for conditional logic
from airflow.operators.python import BranchPythonOperator

def choose_path(**context):
    if context['ds'] == context['ds'][:7] + '-01':  # First of month
        return 'full_refresh'
    return 'incremental_load'

branch = BranchPythonOperator(
    task_id='choose_load_strategy',
    python_callable=choose_path,
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

# dagster_pipeline/assets.py
import pandas as pd
from dagster import asset, AssetIn, Output, MetadataValue
import duckdb

@asset(
    description="Raw orders extracted from PostgreSQL",
    metadata={"source": "production-postgres", "table": "orders"},
)
def raw_orders(context) -> pd.DataFrame:
    """Extract orders from source database."""
    conn = get_postgres_connection()
    df = pd.read_sql("""
        SELECT id, customer_id, amount, status, created_at, updated_at
        FROM orders
        WHERE updated_at >= NOW() - INTERVAL '1 day'
    """, conn)
    context.log.info(f"Extracted {len(df)} orders")
    return df


@asset(
    ins={"raw_orders": AssetIn()},
    description="Cleaned and typed orders",
)
def staged_orders(raw_orders: pd.DataFrame) -> Output[pd.DataFrame]:
    """Clean and normalize orders."""
    df = raw_orders.copy()
    df['amount'] = pd.to_numeric(df['amount'], errors='coerce')
    df['created_at'] = pd.to_datetime(df['created_at'], utc=True)
    df['status'] = df['status'].str.lower().str.strip()
    df = df.dropna(subset=['id', 'amount'])
    df = df[df['amount'] > 0]

    return Output(
        df,
        metadata={
            "num_records": len(df),
            "preview": MetadataValue.md(df.head(5).to_markdown()),
            "schema": MetadataValue.json({col: str(dtype) for col, dtype in df.dtypes.items()}),
        }
    )


@asset(
    ins={"staged_orders": AssetIn(), "staged_customers": AssetIn()},
    description="Customer lifetime value — one row per customer",
)
def customer_lifetime_value(
    staged_orders: pd.DataFrame,
    staged_customers: pd.DataFrame
) -> Output[pd.DataFrame]:
    """Compute CLV by joining orders and customers."""
    joined = staged_orders.merge(staged_customers, on='customer_id', how='left')
    clv = (
        joined
        .query("status == 'completed'")
        .groupby('customer_id')
        .agg(
            total_orders=('id', 'count'),
            lifetime_revenue=('amount', 'sum'),
            avg_order_value=('amount', 'mean'),
            last_order_date=('created_at', 'max'),
        )
        .reset_index()
    )
    return Output(
        clv,
        metadata={
            "num_customers": len(clv),
            "total_revenue": float(clv['lifetime_revenue'].sum()),
        }
    )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

# dagster_pipeline/__init__.py
from dagster import Definitions, ScheduleDefinition, define_asset_job

# Define a job that materializes all assets
daily_pipeline_job = define_asset_job(
    name="daily_pipeline",
    selection="*",    # All assets
)

# Schedule it
daily_schedule = ScheduleDefinition(
    job=daily_pipeline_job,
    cron_schedule="0 2 * * *",  # 2 AM daily
)

defs = Definitions(
    assets=[raw_orders, staged_orders, customer_lifetime_value],
    jobs=[daily_pipeline_job],
    schedules=[daily_schedule],
)

1
2
3
4
5

# Run locally
dagster dev                              # Start dev UI on http://localhost:3000
dagster asset materialize --select "*"  # Materialize all assets
dagster asset materialize --select raw_orders  # Single asset
dagster job execute -j daily_pipeline   # Run the full job