dbt Dimensional Modeling for Streaming Pipelines

When people hear "streaming pipeline," they often assume that means real-time dashboards and no data warehouse. The reality is that most streaming use cases still need a warehouse layer for historical analysis, model training, and reporting. The data just gets there faster.

In AthleteOS, Apache Flink consumes from two Kafka topics, joins streams in real time, and writes to Snowflake every 30 seconds via the Snowflake Kafka connector. dbt then runs on a schedule to transform raw Snowflake tables into clean dimensional models. Here's how I structured that layer.

The raw layer

Flink writes to two raw Snowflake tables:

• raw.pose_events — one row per MediaPipe BlazePose frame: session_id, timestamp_ms, keypoints (JSON), form_score, flagged_errors (array)
• raw.biometric_readings — one row per WHOOP API poll: session_id, timestamp_ms, hrv, strain, recovery_score, heart_rate

These tables are append-only. Flink never updates or deletes — it only inserts. This is an important property to preserve in the dbt layer: I never build models that require row-level updates to the raw tables.

Staging models

The staging layer handles type casting, null handling, and renaming. One staging model per raw source:

-- models/staging/stg_pose_events.sql
select
    session_id,
    to_timestamp(timestamp_ms / 1000) as event_at,
    try_parse_json(keypoints)         as keypoints,
    form_score::float                 as form_score,
    flagged_errors                    as flagged_errors,
    _ingested_at                      as ingested_at
from {{ source('raw', 'pose_events') }}
where session_id is not null

Nothing clever happens in staging. If you're doing business logic in staging, it's in the wrong layer.

Intermediate models

The intermediate layer is where the streaming-specific complexity lives. Because Flink writes every 30 seconds, a single training session produces ~180 pose event records per hour. The intermediate layer aggregates these into session-level metrics:

-- models/intermediate/int_session_pose_metrics.sql
select
    session_id,
    min(event_at)                          as session_start_at,
    max(event_at)                          as session_end_at,
    datediff('minute',
        min(event_at), max(event_at))      as duration_mins,
    count(*)                               as total_readings,
    avg(form_score)                        as avg_form_score,
    min(form_score)                        as min_form_score,
    array_agg(distinct value)              as all_errors
from {{ ref('stg_pose_events') }},
     lateral flatten(input => flagged_errors) f
group by session_id

The key insight: the intermediate layer converts streaming cardinality (one row per 30s) into analytical cardinality (one row per session). Everything downstream works with sessions, not events.

Mart models

The mart layer joins intermediates into the final fact and dimension tables consumed by the application and analytics tools:

-- models/marts/fct_sessions.sql
select
    p.session_id,
    p.session_start_at,
    p.session_end_at,
    p.duration_mins,
    p.avg_form_score,
    p.min_form_score,
    p.all_errors,
    b.avg_hrv,
    b.avg_heart_rate,
    b.max_strain,
    b.entry_recovery_score
from {{ ref('int_session_pose_metrics') }}  p
left join {{ ref('int_session_biometrics') }} b
    using (session_id)

The left join matters here. Some sessions have pose data but no biometric data (WHOOP API was offline). The model should still produce a row for those sessions rather than silently dropping them.

Incremental models for high-cardinality sources

The raw tables grow continuously. Full refreshes become expensive fast. I use incremental models for everything in the staging and intermediate layers:

{{ config(
    materialized='incremental',
    unique_key='session_id',
    on_schema_change='append_new_columns'
) }}

select ...
from {{ source('raw', 'pose_events') }}
{% if is_incremental() %}
    where _ingested_at > (select max(_ingested_at) from {{ this }})
{% endif %}

on_schema_change='append_new_columns' is important for streaming workloads. Flink schema evolution can add new fields to the Kafka topic, and this setting prevents dbt from failing when new columns appear in the source — it just adds them to the model.

Testing

dbt tests catch data quality issues that Flink and the Kafka connector can introduce:

models:
  - name: fct_sessions
    columns:
      - name: session_id
        tests:
          - unique
          - not_null
      - name: avg_form_score
        tests:
          - not_null
          - accepted_range:
              min_value: 0
              max_value: 1
      - name: duration_mins
        tests:
          - not_null
          - accepted_range:
              min_value: 1
              max_value: 300

I run these tests in the Airflow DAG that triggers dbt, downstream of the dbt run step. Failed tests halt downstream jobs — no broken data reaches the application layer.

The principle that held

Kimball's dimensional modeling principles are 30 years old and they hold for streaming data just as well as batch. The only adjustment is the intermediate aggregation step that converts streaming cardinality into analytical cardinality. Everything else — staging → intermediate → mart, incremental loading, testing at mart boundaries — transfers directly.