Incremental ETL and Change Data Capture for ML: Boost Model Freshness and Cut Serving Latency

Why incremental ETL and CDC matter for ML pipelines

Keeping machine learning models useful often depends more on data freshness than on model complexity. Incremental ETL and Change Data Capture (CDC) reduce end to end latency for feature updates, shrink compute cost, and help models adapt to recent behavior. This article walks through practical patterns, trade offs, and concrete code snippets to help data teams move from full batch reloads to efficient incremental flows.

Core ideas in plain terms

Incremental ETL: transfer only new or changed rows since the last run, based on timestamps, log sequence numbers, or watermarks.
CDC: capture every row level change from the source transactional system and stream those events to downstream systems in near real time.
Feature freshness vs cost: more frequent updates improve freshness but increase compute and operational complexity.

When to choose incremental vs CDC

Use incremental ETL when source change volume is moderate and you can rely on a monotonic updated timestamp or an increasing id.
Choose CDC when you need low latency, cannot rely on reliable timestamps, or must reproduce deletes and updates precisely.
Hybrid: run incremental microbatches for analytical aggregates and stream raw CDC events into a feature store for critical features.

Architecture patterns and components

Typical components to combine:

Source DB with reliable updated_at or WAL access (Postgres, MySQL).
CDC connector like Debezium or cloud change streams.
Message bus such as Kafka or cloud pubsub for buffering and fanout.
Stream processor: Spark Structured Streaming, Flink, or ksqlDB for enrichment and reduction.
Feature store or serving DB: Feast, Redis, Cassandra, or a Delta Lake table for batch serving.
Orchestrator: Airflow, Dagster, or a stream deployment for continuous jobs.

Concrete incremental ETL example: Postgres to feature table

This pattern reads only rows changed since last run using an updated_at column. Store the last watermark in metadata and upsert new rows into a data lake or feature table.

import psycopg2
from datetime import datetime, timezone

def load_incremental(last_watermark):
    conn = psycopg2.connect(dbname=\'mydb\', user=\'etl_user\', password=\'secret\')
    cur = conn.cursor()
    cur.execute(
        \"\"\"SELECT id, user_id, feature_a, feature_b, updated_at
           FROM events
           WHERE updated_at > %s
           ORDER BY updated_at ASC
        \"\"\", (last_watermark,)
    )
    rows = cur.fetchall()
    cur.close()
    conn.close()
    return rows

def upsert_to_feature_table(rows, writer):
    # writer can be a delta lake writer, DB upsert, or feature store client
    for r in rows:
        writer.upsert(dict(id=r[0], user_id=r[1], feature_a=r[2], feature_b=r[3], updated_at=r[4]))

if __name__ == \'__main__\':
    last = datetime.fromisoformat(\'2024-01-01T00:00:00+00:00\')
    rows = load_incremental(last)
    if rows:
        # example writer interface
        writer = MyFeatureWriter()
        upsert_to_feature_table(rows, writer)
        new_watermark = rows[-1][4].astimezone(timezone.utc).isoformat()
        persist_watermark(new_watermark)

Key points: persist a reliable watermark, sort by updated_at to avoid reordering issues, and implement idempotent upsert logic at the sink.

CDC flow example: Debezium + Kafka + Spark Structured Streaming

Debezium captures WAL changes and writes them to Kafka topics. A streaming job reads events, applies lightweight enrichment, and merges them into a Delta Lake table for serving.

# pyspark example for merging CDC events into Delta
from pyspark.sql import SparkSession
from delta.tables import DeltaTable

spark = SparkSession.builder.appName(\'cdc_merge\').getOrCreate()

# read Kafka topic with CDC events (Avro or JSON payload)
df = (spark.readStream.format(\'kafka\')
      .option(\'kafka.bootstrap.servers\', \'kafka:9092\')
      .option(\'subscribe\', \'dbserver1.inventory.customers\')
      .load())

# parse value and select required fields
events = (df.selectExpr(\'CAST(value AS STRING) as payload\')
          .selectExpr(\'json_tuple(payload, \\\'after\\\', \\\'op\\\') as (after, op)\')
          .selectExpr(\'json_tuple(after, \\\'id\\\', \\\'user_id\\\', \\\'email\\\') as (id, user_id, email)\'))

# write to a checkpointed streaming sink that performs upserts
def foreach_batch(batch_df, batch_id):
    target = DeltaTable.forPath(spark, \'/mnt/delta/features/customers\')
    batch_df.createOrReplaceTempView(\'staging\')
    # perform merge by id
    (target.alias(\'t\')
       .merge(batch_df.alias(\'s\'), \'t.id = s.id\')
       .whenMatchedUpdateAll()
       .whenNotMatchedInsertAll()
       .execute())

(events.writeStream
 .foreachBatch(foreach_batch)
 .option(\'checkpointLocation\', \'/mnt/checkpoints/cdc_customers\')
 .start())

Streaming merges keep a single source of truth, and Delta Lake handles concurrency for upserts. This reduces serving latency since fresh features are available soon after change commit.

Integrating with a feature store

Feature stores separate online and offline stores. For low latency serving, stream updates to an online store such as Redis or a key value store. For training, keep a historical offline store like a Delta table or Parquet files.

# minimal Feast write example (pseudo)
from feast import FeatureStore
from datetime import datetime

fs = FeatureStore(repo_path=\'/path/to/repo\')

# build a pandas dataframe with new features
import pandas as pd
df = pd.DataFrame([{\'user_id\': 123, \'feature_x\': 0.75, \'ts\': datetime.utcnow()}])

fs.materialize_incremental(end_date=datetime.utcnow())
fs.write_to_online_store(df, entity_keys=[\'user_id\'])

Materialize incremental avoids recomputing full features. When streaming, send CDC processed rows directly to the online write API to minimize serving latency.

Dealing with common pitfalls

Out of order events: use event time and watermarks in streaming engines, and keep idempotent sinks.
Schema evolution: register schema with a registry and handle nullable new fields in enrichment steps.
Deletes and tombstones: CDC provides delete events; ensure sinks implement delete semantics or soft delete flags.
Backpressure and spikes: buffer with Kafka and autoscale stream processing; avoid synchronous downstream writes.

Orchestration patterns

Airflow still works well for incremental jobs and for metadata operations such as materializing features. Streaming jobs run continuously, but an orchestrator can manage deployments, schema migrations, and backfills.

# simplified Airflow DAG snippet for incremental ETL
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

def run_incremental():
    last = read_watermark()
    rows = load_incremental(last)
    if rows:
        write_rows(rows)
        persist_watermark(rows[-1][4].isoformat())

with DAG(dag_id=\'incremental_etl\', start_date=datetime(2024,1,1), schedule_interval=timedelta(minutes=15)) as dag:
    task = PythonOperator(task_id=\'incremental_load\', python_callable=run_incremental)

Schedule depends on freshness needs. Short intervals reduce staleness but increase runs. Monitor lag between source commit time and feature availability to find an operational sweet spot.

Measuring impact on ML metrics and serving

Track model performance drift alongside freshness metrics. Useful signals include:

Feature age distribution at inference time.
End to end latency from source commit to online feature availability.
Model metric changes after improving freshness for a subset of features.

Run A/B experiments where one cohort uses fresher features and compare offline and online metrics. Avoid claiming universal gains; improvements depend on feature volatility and model sensitivity.

Checklist before moving to streaming CDC

Validate source ordering guarantees and transactional semantics.
Design idempotent sink operations for upserts and deletes.
Implement schema registry and migration plan.
Set observability: event lag, processing errors, sink write failures.
Plan backfill and historical reprocessing strategy.

Final practical tips

Start by converting expensive batch jobs to incremental microbatches, measure latency and cost, then adopt CDC for features that require the lowest staleness. Use small, testable components and automate watermark handling. Keep examples reproducible: a small Postgres table, a Kafka topic, and a streaming merge job often suffice to evaluate production readiness.