Deep Dive: Building a Data Platform with CDC, dlt, Snowflake & dbt Core

Retail and manufacturing systems generate a continuous stream of changes: new orders, payment updates, cancellations, and returns. Traditional extraction based on nightly SELECT * FROM sales full-table scans can create contention, repeatedly transfer unchanged data, and delay reporting until the next morning.

1. An End-to-End Medallion Architecture

To address these challenges, I organized the platform into a Medallion Architecture with three distinct layers:

[SQL Server 2022 CDC]  -- (Transaction Log Stream)
          │
          ▼
    [dlt Python Ingest] -- (RSA Key-Pair Auth / Cursor by LSN)
          │
          ▼
 [Snowflake BRONZE]   -- (Append-Only Raw Archive)
          │
          ▼
  [dbt Core STAGING]   -- (CDC Deduplication: ROW_NUMBER over _start_lsn)
          │
          ├─► [dbt SNAPSHOTS] -- (SCD Type 2: snap_products price changes)
          │
          ▼
   [Snowflake GOLD]    -- (Kimball Star Schema: fct_sales merged + dim tables)
          │
          ▼
   [Metabase BI Tool]  -- (12 Business KPIs & Executive Revenue Dashboards)

2. Enabling CDC & Extracting Data with Python and dlt

On Microsoft SQL Server 2022 running in Docker, I enable Change Data Capture (CDC) at both database and transaction-table level:

-- Enable CDC at database level
EXEC sys.sp_cdc_enable_db;

-- Enable CDC for dbo.sales
EXEC sys.sp_cdc_enable_table
  @source_schema = N'dbo',
  @source_name = N'sales',
  @role_name = NULL,
  @supports_net_changes = 1;

With CDC enabled, the capture process reads the transaction log and writes changes to cdc.dbo_sales_CT. Each change carries a log position called a Log Sequence Number (LSN) (in column __$start_lsn) and an operation code __$operation (1 = Delete, 2 = Insert, 3 = Update Before, 4 = Update After).

For ingestion, I use the open-source Python library dlt (data load tool) . I configure an incremental cursor on the binary column __$start_lsn:

import dlt
from dlt.sources.sql_database import sql_database

# Use an incremental LSN cursor to extract new events
source = sql_database(
    credentials="mssql+pyodbc://sa:Password@localhost:1433/AdventureWorks",
    schema="cdc"
)
source.resources["dbo_sales_CT"].apply_hints(
    incremental=dlt.sources.incremental(
        '\"__$start_lsn\"', initial_value=b"\x00" * 10
    )
)

# Append CDC records to the Snowflake Bronze schema
pipeline = dlt.pipeline(
    pipeline_name="adventureworks_cdc_pipeline",
    destination="snowflake", dataset_name="bronze"
)
load_info = pipeline.run(source.with_resources("dbo_sales_CT"), write_disposition="append")
print(load_info)

3. Snowflake RSA Key-Pair Authentication

For automated loading, I use key-pair authentication rather than embedding a static account password in scripts. I configure RSA Key-Pair Authentication with a 2048-bit key for the service account DLT_USER trong Snowflake:

• The public key is assigned to the Snowflake user: ALTER USER DLT_USER SET RSA_PUBLIC_KEY = '...';
• The private key is protected with a passphrase and managed outside source code through the runtime environment.
• Following least privilege, I assign a dedicated role, INGESTION_ROLEwith the permissions needed to create tables and load data in BRONZE.

4. CDC Deduplication in dbt Staging

Append-only CDC ingestion can produce multiple events for the same order as its status changes. Reporting directly on those events would count the same order more than once.

Trong model src_cdc_sales.sql in dbt Core, I deduplicate records with the window function ROW_NUMBER():

WITH raw_cdc_events AS (
  SELECT * FROM {{ source('bronze', 'dbo_sales_ct') }}
), ranked_events AS (
  SELECT order_number, order_line_item, customer_key, product_key,
    order_quantity, unit_price, order_date,
    __$operation AS cdc_operation, __$start_lsn AS cdc_lsn,
    -- Rank the most recent LSN and sequence value first
    ROW_NUMBER() OVER (
      PARTITION BY order_number, order_line_item
      ORDER BY __$start_lsn DESC, __$seqval DESC
    ) AS event_rank
  FROM raw_cdc_events
  -- Exclude UPDATE_BEFORE records
  WHERE __$operation != 3
)
SELECT order_number, order_line_item, customer_key, product_key,
  order_quantity, unit_price, order_date,
  CASE WHEN cdc_operation = 1 THEN TRUE ELSE FALSE END AS is_deleted
FROM ranked_events
WHERE event_rank = 1;

“This logic keeps the latest recorded state for each order line. Delete events set is_deleted so that the fact model can handle them as soft deletes.”

5. Tracking Historical Changes with dbt Snapshots

Product prices and standard costs change over time. Using today's price for a sale from two years ago would distort historical reporting.

I configure dbt Snapshot to implement Slowly Changing Dimension Type 2 for the product table (snap_products):

{% snapshot snap_products %}

{{
    config(
      target_schema='gold',
      unique_key='product_key',
      strategy='check',
      check_cols=['product_price', 'product_cost', 'product_name']
    )
}}

SELECT 
    product_key,
    product_name,
    model_name,
    product_cost,
    product_price
FROM {{ ref('stg_products') }}

{% endsnapshot %}

dbt maintains the columns dbt_valid_from and dbt_valid_to. When joining to Fact Sales, a date-range condition using order_date BETWEEN dbt_valid_from AND COALESCE(dbt_valid_to, '9999-12-31')selects the product version valid at the transaction's date.

6. Kimball Star Schema & Incremental Merges into Fact Sales

In the GOLD schema, the data follows a Kimball Star Schema:

• Fact Table: fct_sales (23,935 order lines). The model uses an incremental_strategy='merge' strategy with the primary key unique_key=['order_number', 'order_line_item']. Each scheduled dbt run processes new or changed orders rather than rebuilding the entire fact table, reducing unnecessary warehouse work.
• Dimension Tables: dim_customers, dim_products, dim_territories, dim_calendar (supporting fiscal-year and quarter analysis).

7. Data Quality with 30 Automated dbt Tests & CI/CD

To keep invalid data out of Gold, I define 30 automated test cases:

• Generic Tests: Check unique and not_null on all primary and surrogate keys.
• Referential Integrity Tests: Check relationships checks ensure each customer_key and product_key in Fact Sales references a valid record in its corresponding dimension.
• Singular Business Logic Tests: Custom SQL tests check that order quantities are not negative (order_quantity > 0) and order dates do not fall in the future.

Conclusion & Open Source

AdventureWorks Modern Data Platform demonstrates my approach to architecture and implementation. The Docker Compose setup, dlt ingestion scripts, dbt models, snapshots, and tests are available on GitHub.

Explore the repository on GitHub (tranduc2204/adventure_works) ↗