cdc-streaming-pipeline

Files (5)

• LICENSE
• SKILL.md
• event-source-mapping.yaml · reference/
• lambda-consumer.py · reference/
• redshift-ddl.sql · reference/

SKILL.md (first lines)

# CDC streaming pipeline (Kinesis -> Redshift Serverless / S3)

This skill captures the lived patterns for building a real-time CDC pipeline:
a transactional source publishes change events to Kinesis, a Lambda consumer
applies them to a sink (Redshift Serverless event log, or S3 / Iceberg via
Firehose). It is source-agnostic: the producer-side notes are split into
"any CDC source" and "Aurora DSQL preview" callouts.

## When to use

- You are wiring up a CDC pipeline from Aurora DSQL, Aurora Postgres, or RDS
  to Kinesis Data Streams (or MSK) with a Lambda consumer.
- You need to write the consumer Lambda that pushes CDC rows into Redshift
  Serverless via the Redshift Data API.
- You are deciding how to model CDC events at the sink (append-only vs
  in-place upsert, how to absorb schema drift).
- You want the operational gotchas (parameter limits, async statement
  polling, retries, poison records) before you debug them.

## When NOT to use

- You only need the lakehouse / cold-path target (S3 Tables, Iceberg,
  Redshift external schema, federated Glue catalogs). Use the
  `firehose-iceberg-pipeline` and `lakehouse-redshift` skills for those.
- You want generic Redshift Serverless setup unrelated to CDC (workgroup
  sizing, RPU pricing, snapshot policy). Out of scope here.
- You are doing batch ETL / DMS full-load. CDC is for ongoing change
  capture; bulk loads belong elsewhere.
- You need EMR / Glue / Spark transforms inside the stream. This skill
  uses Lambda + Redshift Data API.

If you ARE building the cold path on top of this hot path, ALSO load
`firehose-iceberg-pipeline` (it covers the Firehose -> S3 Tables -> Iceberg
side) and `lakehouse-redshift` (it covers the Redshift external schema +
hot/cold UNION view).

## Architecture in one paragraph

Source DB -> CDC stream (DSQL CDC, DMS, Debezium, or an RDS native CDC
mechanism) -> Kinesis Data Stream -> Lambda event source mapping ->
parameterized INSERTs into a Redshift Serverless append-only `cdc_events`
log table -> per-source-table `*_current` views reconstruct current state
via `ROW_NUMBER() OVER (PARTITION BY pk ORDER BY commit_ts DESC)`.

## 1. The append-only event log + reconstruction pattern

This is the single most important design choice. Use it.

### Why append-only

CDC delivery is **unordered and may duplicate**. Two facts together kill
naive `INSERT ... ON CONFLICT ... UPDATE` upserts:

1. Records can arrive out of commit order (especially across Kinesis
   shards, but also within a shard during retries).
2. The same record can be delivered more than once (Lambda batch retry,
   producer retry, BisectBatchOnFunctionError).

Append-only writes are idempotent under both. Every CDC event becomes a
new row; current state is reconstructed at read time by picking the
latest commit timestamp per primary key. Late-arriving older events are
discarded by the window function, not by re-doing an UPDATE.

### `cdc_events` table DDL (Redshift Serverless)

Generic shape that absorbs any source table by routing all events to a
single SUPER column:

```sql
CREATE TABLE IF NOT EXISTS cdc_events (
    event_id          BIGINT IDENTITY(1,1) PRIMARY KEY,
    source_table      VARCHAR(100) NOT NULL,
    operation         VARCHAR(10)  NOT NULL,    -- "c" or "d"
    record_id         VARCHAR(50)  NOT NULL,    -- source row primary key
    event_data        SUPER,                     -- full row state
    commit_timestamp  TIMESTAMP    NOT NULL,    -- source-side commit time
    ingested_at       TIMESTAMP    NOT NULL DEFAULT GETDATE()
)
DISTSTYLE KEY
DISTKEY (record_id)
SORTKEY (source_table, commit_timestamp);

-- Lambda's IAM-mapped DB user is auto-created on first GetCredentials.
-- Granting to PUBLIC keeps the demo simple. In production, grant to a
-- specific role matching the IAM identity (typically "IAMR:<role-name>")
-- and drop the PUBLIC grant.
GRANT INSERT, SELECT ON cdc_events TO PUBLIC;
```

Why these choices:

- `BIGINT IDENTITY` gives a stable event ordering for debugging without
  participating in correctness (correctness comes from `commit_timestamp`).
- `record_id VARCHAR(50)`: stringify all PKs at the consumer. UUIDs,
  bigints, composite keys all serialize cleanly.
- `event_data SUPER`: lets a single table absorb any source schema. New
  source tables work with zero DDL changes.
- `DISTKEY(record_id)` colocates all events for a given row on the same
  slice, so the `ROW_NUMBER` window in current-state views runs locally.
- `SORTKEY(source_table, commit_timestamp)`: zone maps prune by table
  and time on every per-table view query.

### Current-state view DDL (Redshift Serverless flavor with SUPER)

One view per source table. `event_data."col"::TYPE` is the SUPER subscript
+ cast pattern. (Use the SQL standard cast `::`; `CAST(... AS TYPE)` works
too but is verbose.)

```sql
CREATE OR REPLACE VIEW orders_current AS
SELECT
    record_id                            AS order_id,
    event_data."customer_id"::VARCHAR    AS customer_id,
    event_data."total_cents"::BIGINT     AS total_cents,
    event_data."status"::VARCHAR         AS status,
    commit_timestamp                     AS last_change_at
FROM (
    SELECT *,
           ROW_NUMBER() OVER (

From Kilo-Org/kilo-marketplace, Apache-2.0.

More from Kilo-Org/kilo-marketplace