--- name: cdc-streaming-pipeline description: Build a real-time change-data-capture pipeline from a transactional database (Aurora DSQL, Aurora Postgres, RDS) to Redshift Serverless, S3, or any other sink, via Kinesis Data Streams (or MSK). Covers producer setup, consumer Lambda correctness, the append-only event log + reconstruction pattern, schema-drift handling with SUPER + JSON_PARSE, and the operational gotchas (Redshift Data API parameter caps, async statement polling, poison-record handling, retry-and-bisect). Use when the user asks for "CDC to Redshift", "stream change data capture", "Aurora CDC", "DSQL CDC", "Postgres replication to S3", or "build a Lambda consumer for Kinesis CDC". Aurora DSQL public-preview specifics are flagged; the rest is source-agnostic. --- # 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:") -- 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 ( PARTITION BY record_id ORDER BY commit_timestamp DESC ) AS rn FROM cdc_events WHERE source_table = 'orders' ) WHERE rn = 1 AND operation <> 'd'; ``` Key points: - Filter `source_table` BEFORE the window function (the planner pushes it down; the explicit predicate also keeps zone-map pruning effective). - `WHERE operation <> 'd'` is the tombstone. Deletes still produce the latest event for that PK; the view just hides them. - The cast `event_data."col"::TYPE` extracts a SUPER field. Quote the field name; case-sensitive. ### Aurora DSQL preview gotcha: only `c` and `d` ops In the DSQL CDC public preview, the `op` field on every event is one of: - `c` for CREATE (both inserts and updates arrive as `c`) - `d` for DELETE There is **no `u` op**. The reconstruction pattern handles this correctly because the latest `c` per `record_id` wins regardless of whether the source operation was an insert or an update. Don't write code that special- cases `op == 'u'`; it will never fire on DSQL today. For non-DSQL sources (Debezium, DMS), expect `c`, `u`, `d` (and possibly `r` for snapshot reads). The same `ROW_NUMBER` reconstruction works as long as you record `commit_timestamp` and treat any of `c | u` as "row exists with this state". ## 2. Producer-side gotchas ### Aurora DSQL preview: no CFN resource type for the CDC stream DSQL CDC is in public preview and does NOT yet have a CloudFormation resource type. You cannot put the CDC stream in a CFN template. Create it via the AWS CLI (or SDK) AFTER the cluster + Kinesis stream + IAM role exist: ```bash aws dsql create-stream \ --cluster-identifier "${DSQL_CLUSTER_ID}" \ --target-definition "$(printf '{"kinesis":{"streamArn":"%s","roleArn":"%s"}}' \ "${KINESIS_STREAM_ARN}" "${DSQL_CDC_ROLE_ARN}")" \ --ordering UNORDERED \ --format JSON \ --region "${AWS_REGION}" ``` Make the script idempotent. Reuse an existing stream rather than failing: ```bash existing=$(aws dsql list-streams \ --cluster-identifier "${DSQL_CLUSTER_ID}" \ --region "${AWS_REGION}" \ --query 'streams[0].streamIdentifier' \ --output text 2>/dev/null || true) if [ -n "${existing}" ] && [ "${existing}" != "None" ]; then DSQL_STREAM_ID="${existing}" else DSQL_STREAM_ID=$(aws dsql create-stream ... --query 'streamIdentifier' --output text) fi # Wait for ACTIVE before continuing. for i in $(seq 1 60); do status=$(aws dsql get-stream \ --cluster-identifier "${DSQL_CLUSTER_ID}" \ --stream-identifier "${DSQL_STREAM_ID}" \ --region "${AWS_REGION}" \ --query 'status' --output text) case "${status}" in ACTIVE) break ;; FAILED|DELETING) echo "stream entered ${status}" >&2; exit 1 ;; *) sleep 5 ;; esac done ``` The `--ordering UNORDERED` is the documented default. Picking `UNORDERED` is correct for the append-only pattern; it also gives DSQL the most freedom to parallelize across shards. ### Trust policy for the DSQL-to-Kinesis role The CDC stream assumes an IAM role to put records to your Kinesis stream. That role must trust the `dsql.amazonaws.com` service principal AND the condition keys must include the cluster's resource ARN so a different DSQL cluster cannot impersonate yours: ```yaml DsqlCdcKinesisRole: Type: AWS::IAM::Role Properties: AssumeRolePolicyDocument: Version: "2012-10-17" Statement: - Effect: Allow Principal: Service: dsql.amazonaws.com Action: sts:AssumeRole Condition: StringEquals: "aws:SourceAccount": !Ref AWS::AccountId ArnLike: # DSQL passes the stream ARN (which lives under the cluster) # as aws:SourceArn, NOT the bare cluster ARN. Use ArnLike # with the /stream/* suffix so the condition actually matches. "aws:SourceArn": !Sub "${DsqlCluster.ResourceArn}/stream/*" Policies: - PolicyName: PutCdcToKinesis PolicyDocument: Version: "2012-10-17" Statement: - Effect: Allow Action: - kinesis:PutRecord - kinesis:PutRecords - kinesis:DescribeStreamSummary - kinesis:ListShards Resource: !GetAtt CdcStream.Arn ``` ### Schema-drift handling: SUPER (or string) at the sink Source schemas evolve. Two viable strategies: 1. **SUPER column at the sink (recommended).** The Lambda dumps the full row payload as JSON into `event_data SUPER`. Downstream views read specific fields with `event_data."col"::TYPE`. New columns appear automatically; missing columns return NULL when projected. 2. **Stringly-typed VARCHAR(MAX) column.** Same idea, but you `JSON_PARSE` at view time. SUPER is preferable: Redshift Serverless plans queries over SUPER more efficiently, and the field-extraction syntax is shorter. What this strategy buys you: - Adding a new source table is a producer-side change only; no Redshift DDL required for ingestion. You add a new `*_current` view at your leisure. - Adding a column to an existing source table requires no Redshift DDL. Only the affected `*_current` view needs the new column added if you want to surface it. ## 3. Lambda consumer correctness patterns These are non-negotiable. Each one has a "this is what breaks if you skip it" footnote because each one was earned by debugging. ### 3.1 Parameterized SQL only. Never string-concatenate. The Redshift Data API supports named parameters. Use them. The CDC payload contains user-supplied data from the source DB; concatenating it into SQL is unsafe and wrong, even within an internal pipeline. ```python def _build_parameterized_insert(rows: list[dict]) -> tuple[str, list[dict]]: value_clauses = [] parameters = [] for i, r in enumerate(rows): # CAST(:ts AS BIGINT) is required: the Data API type-infers numeric # parameters as INTEGER, but ms-since-epoch (13 digits) overflows # INT4. Dividing by 1000.0 (not 1000) preserves sub-second precision # in commit_timestamp; integer division would truncate to whole seconds. value_clauses.append( f"(:t{i}, :op{i}, :id{i}, JSON_PARSE(:d{i}), " f"TIMESTAMP 'epoch' + CAST(:ts{i} AS BIGINT) / 1000.0 * INTERVAL '1 second')" ) parameters.extend([ {"name": f"t{i}", "value": r["table"]}, {"name": f"op{i}", "value": r["op"]}, {"name": f"id{i}", "value": str(r["record_id"])}, {"name": f"d{i}", "value": json.dumps(r["row"])}, {"name": f"ts{i}", "value": str(r["commit_ts_ms"])}, ]) sql = ( "INSERT INTO cdc_events " "(source_table, operation, record_id, event_data, commit_timestamp) " f"VALUES {', '.join(value_clauses)}" ) return sql, parameters ``` Two non-obvious bits: - `JSON_PARSE(:d{i})` converts the JSON string parameter into a SUPER value at insert time. Without `JSON_PARSE`, the value would be stored as plain text and SUPER subscripting would fail. - `CAST(:ts{i} AS BIGINT)` is required. The Data API looks at parameter values and chooses INT4 by default; 13-digit millisecond timestamps silently overflow. ### 3.2 Redshift Data API has a 200-parameter cap. Chunk to stay under it. The Redshift Data API rejects a single `execute_statement` with more than 200 parameters. With 5 parameters per CDC row, that means at most 40 rows per statement. ```python PARAMS_PER_ROW = 5 MAX_PARAMS_PER_STATEMENT = 200 ROWS_PER_CHUNK = int(os.environ.get( "ROWS_PER_CHUNK", MAX_PARAMS_PER_STATEMENT // PARAMS_PER_ROW )) # 40 ``` When you raise the per-row parameter count (e.g., adding columns to the INSERT), `ROWS_PER_CHUNK` MUST come down accordingly. Don't override `ROWS_PER_CHUNK` higher than `200 // PARAMS_PER_ROW` without revisiting the math. If your Kinesis event source mapping has `BatchSize: 100`, the Lambda receives up to 100 records per invocation. With `ROWS_PER_CHUNK=40`, that's at most 3 chunks (40 + 40 + 20). The Lambda timeout must cover the worst case: `ceil(BatchSize / ROWS_PER_CHUNK) * STATEMENT_POLL_TIMEOUT_S` plus boto3 overhead. ### 3.3 `execute_statement` is async. ALWAYS poll `describe_statement`. This is the single most dangerous footgun. `redshift-data:ExecuteStatement` returns a statement ID immediately; the SQL has not run yet. If you don't poll for completion: - A FAILED statement looks like a SUCCESS to the Lambda. - The Lambda returns 200 OK to Kinesis. - Kinesis advances the iterator past records that never landed in Redshift. - You discover the data loss days later when current-state views are missing rows. The poll loop: ```python def _await_statement(statement_id: str) -> None: deadline = time.monotonic() + STATEMENT_POLL_TIMEOUT_S delay = 0.2 while True: resp = redshift.describe_statement(Id=statement_id) status = resp.get("Status") if status == "FINISHED": return if status in ("FAILED", "ABORTED"): raise RuntimeError( f"Redshift statement {statement_id} ended in {status}: " f"{resp.get('Error', '')}" ) if time.monotonic() > deadline: raise RuntimeError( f"Redshift statement {statement_id} did not finish within " f"{STATEMENT_POLL_TIMEOUT_S}s (last status={status})" ) time.sleep(delay) delay = min(delay * 2, 1.0) # exponential backoff with cap ``` Exponential backoff caps at 1s so a long-running statement doesn't waste Lambda execution budget on sleeps. ### 3.4 Re-raise on failure. Let Kinesis retry. When a chunk fails, RAISE. Do not catch and return. The Kinesis event source mapping treats a Lambda exception as a batch retry signal: ```python for chunk_index, chunk in enumerate(chunks): sql, parameters = _build_parameterized_insert(chunk) try: response = redshift.execute_statement( WorkgroupName=WORKGROUP, Database=DATABASE, Sql=sql, Parameters=parameters, ) statement_ids.append(response["Id"]) _await_statement(response["Id"]) except Exception: logger.exception( "Chunk %d/%d failed (already-submitted statement_ids=%s)", chunk_index + 1, len(chunks), statement_ids, ) raise # Kinesis retries the batch. ``` Logging the already-submitted statement IDs before re-raising is important: on retry, those chunks will be inserted AGAIN. The append-only pattern absorbs that duplication safely (later reconstruction picks the latest by `commit_timestamp`), but operators reading CloudWatch logs need to know what already landed. ### 3.5 Configure `BisectBatchOnFunctionError` and bounded retries On the EventSourceMapping: ```yaml CdcEventSource: Type: AWS::Lambda::EventSourceMapping Properties: EventSourceArn: !GetAtt CdcStream.Arn FunctionName: !Ref CdcProcessorFunction StartingPosition: LATEST BatchSize: 100 MaximumBatchingWindowInSeconds: 5 MaximumRetryAttempts: 5 BisectBatchOnFunctionError: true Enabled: true ``` What `BisectBatchOnFunctionError: true` buys you: when a batch fails, the event source mapping retries with each half of the batch separately. A single poison record gets isolated quickly instead of blocking the entire batch through `MaximumRetryAttempts` retries. What `MaximumRetryAttempts: 5` buys you: a record that consistently fails (truly poisonous, not a transient Redshift hiccup) doesn't block the shard forever. After 5 attempts the iterator advances. Pair this with a Dead Letter Queue (`DestinationConfig.OnFailure`) if you want to inspect what fell off. ### 3.6 Poison-record handling Poison records: malformed payloads (missing `op`, missing PK, missing timestamp, undecodable Base64). Handle them BEFORE submitting SQL. A poison record should be: 1. Logged at WARNING with enough context to identify it (record key, approximate position). 2. Counted as `skipped`. 3. Not added to the SQL submission list. Concretely: ```python def lambda_handler(event, context): rows = [] skipped = 0 for record in event.get("Records", []): try: raw = base64.b64decode(record["kinesis"]["data"]) payload = json.loads(raw) except (KeyError, TypeError, ValueError, binascii.Error) as e: logger.error("Failed to decode record: %s", e) skipped += 1 continue op = payload.get("op") # DSQL preview: only "c" and "d" ops. For non-DSQL sources accept "u" too. if op == "c": row = payload.get("after") elif op == "d": row = payload.get("before") else: logger.warning("Skipping unknown op: %s", op) skipped += 1 continue if not row: logger.warning("Skipping op=%s with empty row payload", op) skipped += 1 continue record_id = row.get("id") ts_ms = payload.get("ts_ms") if record_id is None or ts_ms is None: logger.warning( "Skipping op=%s payload with missing id=%r or ts_ms=%r", op, record_id, ts_ms, ) skipped += 1 continue rows.append({ "table": payload.get("source", {}).get("table", "unknown"), "op": op, "record_id": record_id, "row": row, "commit_ts_ms": ts_ms, }) # ... chunk + execute + await ... ``` Why not raise on poison records: raising would force Kinesis to retry the ENTIRE batch including the well-formed records, then bisect, ad infinitum until `MaximumRetryAttempts` exhausts. The poison record wedges the shard for minutes. Counting+skipping lets the well-formed records land. The trade-off: skipped records are lost unless you also write them to an audit S3 bucket or DLQ. That is acceptable for most CDC pipelines because truly malformed source data is rare and almost always indicates a producer bug, not a data-loss event. ## 4. IAM for the consumer Lambda (Redshift Serverless flavor) The Lambda needs three things: ```yaml LambdaExecRole: Type: AWS::IAM::Role Properties: AssumeRolePolicyDocument: Statement: - Effect: Allow Principal: { Service: lambda.amazonaws.com } Action: sts:AssumeRole ManagedPolicyArns: # Reads from Kinesis + writes CloudWatch logs. - arn:aws:iam::aws:policy/service-role/AWSLambdaKinesisExecutionRole - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole Policies: - PolicyName: RedshiftDataAPI PolicyDocument: Statement: # Redshift Data API actions do NOT support resource-level # permissions. Resource: "*" is the only valid value. - Effect: Allow Action: - redshift-data:ExecuteStatement - redshift-data:DescribeStatement - redshift-data:GetStatementResult Resource: "*" # IAM auth into the Serverless workgroup, scoped to the # workgroup ARN. - Effect: Allow Action: - redshift-serverless:GetCredentials Resource: !GetAtt RedshiftWorkgroup.Workgroup.WorkgroupArn ``` Two specifics worth memorizing: - **Redshift Data API actions don't support resource-level permissions.** Trying `Resource: !Sub "arn:aws:redshift-data:..."` will be rejected. `Resource: "*"` is the only valid value. - **`redshift-serverless:GetCredentials`** is what gives the Lambda workgroup-level auth. The Data API requires either this for IAM auth or a Secrets Manager secret ARN. Prefer IAM auth; less to rotate. ## 5. Schema mapping: SUPER + JSON_PARSE in detail When the Lambda inserts into `event_data SUPER`, the value passed to the Data API is a JSON string. `JSON_PARSE` converts it to a SUPER value at insert time: ```sql -- INSERT side INSERT INTO cdc_events (source_table, operation, record_id, event_data, commit_timestamp) VALUES (:t0, :op0, :id0, JSON_PARSE(:d0), TIMESTAMP 'epoch' + ...); -- SELECT side: subscript + cast SELECT event_data."customer_id"::VARCHAR AS customer_id, event_data."total_cents"::BIGINT AS total_cents FROM cdc_events WHERE source_table = 'orders'; ``` A few rules of thumb: - Quote field names: `event_data."customer_id"`. Without quotes, Redshift lowercases the identifier. Source columns are usually case-sensitive in JSON, so quoting is the defensive default. - Cast at projection time: `::VARCHAR`, `::BIGINT`, `::INT`, `::BOOLEAN`, `::TIMESTAMP`. Without casts you get SUPER values, which most BI tools cannot consume directly. - Nested fields: `event_data."address"."city"::VARCHAR`. Deeper subscripting works. ### Unnesting SUPER arrays with PartiQL Redshift's PartiQL extension lets you treat a SUPER array as a virtual table on the right-hand side of `FROM`, joining it row-by-element with its parent. The pattern is to comma-cross-join the source row with its array field and bind a per-element alias: ```sql -- Suppose event_data has a "tags" array, e.g. -- event_data = {"id": "...", "tags": ["vip","new"]} SELECT e.record_id, e.commit_timestamp, t::VARCHAR AS tag FROM cdc_events AS e, e.event_data."tags" AS t WHERE e.source_table = 'orders'; ``` Output: one row per (event, tag) pair. The `t::VARCHAR` cast produces a scalar; without it `t` stays SUPER. For arrays of objects: ```sql -- event_data = {"id": "...", "items": [{"sku":"A","qty":2}, ...]} SELECT e.record_id, item."sku"::VARCHAR AS sku, item."qty"::INT AS qty FROM cdc_events AS e, e.event_data."items" AS item WHERE e.source_table = 'orders'; ``` Two requirements specific to PartiQL on Redshift: - The session must enable case-sensitive identifiers when subscripting mixed-case JSON keys: `SET enable_case_sensitive_identifier TO TRUE;` (set this once per session, or use unquoted lowercase keys only). - PartiQL unnesting works inside views too. Combine it with `WITH NO SCHEMA BINDING` if the view also references external Iceberg tables; see the `lakehouse-redshift` skill. If you only need a single element by index, no unnest required: `event_data."tags"[0]::VARCHAR`. ### When NOT to use SUPER If your source schema is fixed and small, plain typed columns are simpler and faster to query. SUPER's value comes from schema-drift absorption; if there is no drift, you are paying SUPER's small overhead for nothing. ## 6. Cross-references - **Cold path / lakehouse target.** If you are also archiving CDC events to S3 Tables (Iceberg) for long retention or larger time-window analytics, ALSO load `firehose-iceberg-pipeline` (it covers Firehose with `IcebergDestinationConfiguration`, the transform Lambda you'll need because Firehose maps top-level JSON keys to Iceberg columns by name, and the two-phase CFN deploy required because Firehose validates the destination Iceberg table synchronously at create time). - **Querying hot+cold together from Redshift Serverless.** ALSO load `lakehouse-redshift` for the external schema setup, the LF access-control mode requirement, the bucket-nested federated catalog ARN format, and the `WITH NO SCHEMA BINDING` view trick. - **DSQL specifics beyond CDC.** If the user is also writing migrations or DDL for the source DSQL cluster, the `dsql` skill (from the `databases-on-aws` plugin) covers DSQL-compatible SQL and `dsql_lint`. ## Reference material - `reference/event-source-mapping.yaml` - full CFN snippet for the Kinesis EventSourceMapping with all the safety knobs set. - `reference/lambda-consumer.py` - a complete, working consumer Lambda that implements every pattern above. - `reference/redshift-ddl.sql` - the `cdc_events` table + a worked `*_current` view example with grants.