Connect

aws_dynamodb_cdc

License: This component requires an enterprise license. You can either upgrade to an Enterprise Edition license, or generate a trial license key that's valid for 30 days.

Reads change data capture (CDC) events from DynamoDB Streams.

Introduced in version 4.79.0.

Consumes records from DynamoDB Streams with automatic checkpointing and shard management.

DynamoDB Streams capture item-level changes in DynamoDB tables. This input supports:

Automatic shard discovery and management
Checkpoint-based resumption after restarts
Concurrent processing of multiple shards
Optional initial snapshot of existing table data
Multi-table streaming with auto-discovery by tags or explicit table lists

Table discovery modes

This input supports three table discovery modes:

single (default) - Stream from the tables specified in the tables field
tag - Auto-discover and stream from multiple tables based on DynamoDB table tags. Use table_tag_filter to filter tables (for example, key:value)
includelist - Stream from an explicit list of tables specified in the tables field. This mode is kept for backward compatibility. Use single instead.

When the input streams from more than one table, it reads from all matching tables simultaneously. Each table maintains its own checkpoint state. In tag or includelist mode, use table_discovery_interval to periodically rescan for new tables (useful for dynamically tagged tables).

Prerequisites

The source DynamoDB table(s) must have DynamoDB Streams enabled. You can enable streams with one of these view types:

KEYS_ONLY - Only the key attributes of the modified item
NEW_IMAGE - The entire item as it appears after the modification
OLD_IMAGE - The entire item as it appeared before the modification
NEW_AND_OLD_IMAGES - Both the new and old item images

Snapshots

When snapshot_mode is set to snapshot_only or snapshot_and_cdc, the input first scans the entire table before (or instead of) streaming changes. This is useful for:

Building a replica or cache with all existing data
Syncing historical data to a data warehouse
Populating a search index with existing records

Snapshots are supported only when streaming from a single table.

Snapshots use the DynamoDB Scan API which consumes read capacity units (RCUs). For large tables, this can be expensive and take considerable time. Use snapshot_segments and snapshot_throttle to control RCU consumption.

Snapshots use eventually consistent reads and do not provide point-in-time consistency. Records modified during the snapshot may appear in both the snapshot and CDC stream (with different values). Use snapshot_deduplicate to minimize duplicates.

Checkpointing

Checkpoints are stored in a separate DynamoDB table (configured via checkpoint_table). This table is created automatically if it does not exist. On restart, the input resumes from the last checkpointed position for each shard. Snapshot progress is also checkpointed, allowing resumption mid-snapshot after failures. The principal needs dynamodb:UpdateItem on the checkpoint table, in addition to the read and write permissions it already uses.

A closed shard that has been read to the end is marked completed in the checkpoint table and is skipped after a restart. A child shard is read only after its parent has been read to the end, so changes to an item are emitted in order across shard splits. There are three exceptions, where a child shard is read without waiting and an older change to an item can be emitted after a newer one: the first discovery under start_from: latest, where every shard starts at its latest position; a parent shard that fails to prepare on 5 consecutive shard refresh cycles (a warning is logged); and a parent shard the stream reports as not found. A child shard that keeps waiting for its parent is reported with a warning every 20 shard refresh cycles. If a stored checkpoint points at stream data that has already been trimmed (older than the 24 hour retention), the shard resumes from TRIM_HORIZON and a warning is logged.

Without a namespace, checkpoints are keyed only by stream and shard (or by source table and shard when global_table is enabled). Multiple pipelines that read the same stream and share one checkpoint_table without a namespace overwrite each other’s positions, which causes skipped or duplicated events.

Multiple independent pipelines can share a single checkpoint table by giving each one a distinct checkpoint_namespace (for example one namespace per developer or environment). Namespaces isolate checkpoints from each other: a pipeline only sees checkpoints written under its own namespace, so changing (or removing) the namespace causes the pipeline to restart from start_from. Namespaces do not coordinate consumers: two pipelines sharing the same namespace still overwrite each other’s checkpoints.

The namespace is prefixed to the checkpoint key. When a pipeline restarts under a namespace that has no checkpoints yet, it starts from start_from. With start_from: trim_horizon, this replays all records still available in the stream’s 24-hour retention window. Switching back to a previously used namespace resumes from that namespace’s last checkpoints.

Alternative

For better performance and longer retention (up to 1 year vs 24 hours), consider using Kinesis Data Streams for DynamoDB with the aws_kinesis input instead.

Message format

Each change event is a JSON object with the fields tableName, eventID, eventName (INSERT, MODIFY, or REMOVE), eventVersion, eventSource, and awsRegion, plus a dynamodb object. The dynamodb object contains keys, newImage, and oldImage when the stream view type provides them, along with sequenceNumber, streamViewType, and sizeBytes. Snapshot records have an eventName of READ and carry the whole item in dynamodb.newImage.

Attribute values are converted from DynamoDB’s typed format, such as {"S": "value"}, to plain values, such as "value". Number attributes stay strings, so convert them in a mapping if you need numeric values.

Metadata

This input adds the following metadata fields to each message:

dynamodb_shard_id - The shard ID from which the record was read (empty for snapshot records)
dynamodb_sequence_number - The sequence number of the record in the stream (empty for snapshot records)
dynamodb_approximate_creation_time - RFC3339 approximate creation time of the stream record (empty for snapshot records)
dynamodb_event_name - The type of change: INSERT, MODIFY, REMOVE, or READ (for snapshot records)
dynamodb_table - The name of the DynamoDB table

Metrics

This input emits the following metrics:

dynamodb_cdc_shards_tracked - Total number of shards being tracked (gauge)
dynamodb_cdc_shards_active - Number of shards currently being read from (gauge)
dynamodb_cdc_snapshot_state - Snapshot state: 0=not_started, 1=in_progress, 2=complete (gauge)
dynamodb_cdc_snapshot_records_read - Total records read during snapshot (counter)
dynamodb_cdc_snapshot_segments_active - Number of active snapshot scan segments (gauge)
dynamodb_cdc_snapshot_buffer_overflow - Incremented when the deduplication buffer exceeds its size limit, disabling dedup (counter)
dynamodb_cdc_snapshot_segment_duration - Time taken by each snapshot scan segment to complete (timer)
dynamodb_cdc_checkpoint_failures - Number of failed checkpoint writes to the checkpoint table (counter)
dynamodb_cdc_failover_skipped - Records skipped during global-table failover replay because they predate the resumed cutoff (counter)

Global table checkpoints (multi-region failover)

In active/active or active/passive multi-region deployments, set global_table: true and list the other regions in global_table_replicas so the auto-created checkpoint table is provisioned as a DynamoDB Global Table (v2). Checkpoints then replicate across regions: a failed-over pipeline resumes near the last committed position instead of replaying the whole stream. Because each region’s stream has its own sequence numbers, cross-region resume is time-based (at-least-once, replaying from the trim horizon up to the last replicated record time); same-region restarts still resume exactly. If the checkpoint table already exists, enabling global_table reconciles it towards the desired configuration: any missing replica regions are added via UpdateTable. The existing table must have been created in global mode (it must use a TableId hash key); pointing global_table at a pre-existing non-global checkpoint table fails fast with a clear error rather than mutating it.

When global_table is enabled the principal additionally needs dynamodb:CreateTable, dynamodb:UpdateTable, dynamodb:DescribeTable, dynamodb:DescribeLimits, iam:CreateServiceLinkedRole, and create/describe permissions in each replica region.

Use this reference to:

Look up configuration options for DynamoDB CDC streaming
Find metadata fields available for message processing
Identify checkpointing and performance tuning settings

Common
Advanced

input:
  label: ""
  aws_dynamodb_cdc:
    tables: []
    checkpoint_table: redpanda_dynamodb_checkpoints
    checkpoint_namespace: ""
    start_from: trim_horizon
    auto_replay_nacks: true
    snapshot_mode: none

input:
  label: ""
  aws_dynamodb_cdc:
    tables: []
    table_discovery_mode: single
    table_tag_filter: ""
    table_discovery_interval: 5m
    checkpoint_table: redpanda_dynamodb_checkpoints
    checkpoint_namespace: ""
    global_table: false
    global_table_replicas: []
    batch_size: 1000
    poll_interval: 1s
    start_from: trim_horizon
    checkpoint_limit: 1000
    max_tracked_shards: 10000
    auto_replay_nacks: true
    throttle_backoff: 100ms
    snapshot_mode: none
    snapshot_segments: 1
    snapshot_batch_size: 100
    snapshot_throttle: 100ms
    snapshot_deduplicate: true
    snapshot_buffer_size: 100000
    region: "" # No default (optional)
    endpoint: "" # No default (optional)
    tcp:
      connect_timeout: 0s
      keep_alive:
        idle: 15s
        interval: 15s
        count: 9
      tcp_user_timeout: 0s
    credentials:
      profile: "" # No default (optional)
      id: "" # No default (optional)
      secret: "" # No default (optional)
      token: "" # No default (optional)
      from_ec2_role: false # No default (optional)
      role: "" # No default (optional)
      role_external_id: "" # No default (optional)

Message structure

Each CDC event is delivered as a JSON message with the following structure. Use these fields in your Bloblang mappings with this.<field>:

{
  "eventID": "abc123-",                    (1)
  "eventName": "INSERT | MODIFY | REMOVE",   (2)
  "eventSource": "aws:dynamodb",
  "awsRegion": "us-east-1",
  "tableName": "my-table",                   (3)
  "dynamodb": {
    "keys": {                                (4)
      "pk": "user#123",
      "sk": "profile"
    },
    "newImage": {                            (5)
      "pk": "user#123",
      "sk": "profile",
      "name": "Alice",
      "email": "alice@example.com"
    },
    "oldImage": {                            (6)
      "pk": "user#123",
      "sk": "profile",
      "name": "Alice Smith"
    },
    "sequenceNumber": "12345678901234567890", (7)
    "sizeBytes": 256,
    "streamViewType": "NEW_AND_OLD_IMAGES"
  }
}

1	Unique identifier for this change event.
2	Type of change: `INSERT` (new item), `MODIFY` (updated item), or `REMOVE` (deleted item).
3	Name of the source DynamoDB table.
4	Primary key attributes of the changed item. Always present.
5	Item state after the change. Present for `INSERT` and `MODIFY` events (requires `NEW_IMAGE` or `NEW_AND_OLD_IMAGES` stream view type).
6	Item state before the change. Present for `MODIFY` and `REMOVE` events (requires `OLD_IMAGE` or `NEW_AND_OLD_IMAGES` stream view type).
7	Position of this record in the shard, used for ordering and checkpointing.

Example mapping

pipeline:
  processors:
    - mapping: |
        root.event_type = this.eventName
        root.table = this.tableName
        root.keys = this.dynamodb.keys
        root.new_data = this.dynamodb.newImage
        root.old_data = this.dynamodb.oldImage

Fields

`auto_replay_nacks`

Whether to automatically replay rejected messages (negative acknowledgements, or nacks) at the output level. If the cause of rejections persists, leaving this option enabled can result in back pressure.

Set auto_replay_nacks to false to delete rejected messages. Disabling auto replays can greatly improve memory efficiency of high throughput streams, as the original shape of the data is discarded immediately upon consumption and mutation.

Requires version 4.106.0 or later.

Type: bool

Default: true

`batch_size`

Maximum number of records to read per shard in a single request. Valid range: 1-1000.

Type: int

Default: 1000

`checkpoint_limit`

Maximum number of unacknowledged messages before forcing a checkpoint update. Lower values provide better recovery guarantees but increase write overhead.

Type: int

Default: 1000

`checkpoint_namespace`

Isolates this pipeline’s checkpoints within a shared checkpoint_table by prefixing the namespace to the checkpoint key. Use this so that multiple pipelines reading the same stream can share one checkpoint table without overwriting each other’s positions, for example per-developer or per-environment test pipelines. Leave empty (the default) to keep the original checkpoint keys unchanged. A namespace isolates readers but does not coordinate them: pipelines that share the same namespace still collide. Changing or removing the namespace changes the checkpoint key. If no checkpoints exist yet under the new key, the pipeline starts from start_from. Switching back to a previously used namespace resumes from that namespace’s last checkpoints. The value cannot contain a # character.

Requires version 4.101.0 or later.

Type: string

Default: ""

`checkpoint_table`

DynamoDB table name for storing checkpoints. Will be created if it doesn’t exist.

Type: string

Default: redpanda_dynamodb_checkpoints

`credentials`

Manually configure the AWS credentials to use (optional). For more information, see the Amazon Web Services guide.

Type: object

`credentials.from_ec2_role`

Use the credentials of a host EC2 machine configured to assume an IAM role associated with the instance.

Type: bool

`credentials.id`

The ID of the AWS credentials to use.

Type: string

`credentials.profile`

The profile from ~/.aws/credentials to use.

Type: string

`credentials.role`

The ARN of the role to assume.

Type: string

`credentials.role_external_id`

An external ID to use when assuming a role.

Type: string

`credentials.secret`

The secret for the AWS credentials in use.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Secrets.

Type: string

`credentials.token`

The token for the AWS credentials in use. Required only when using short-term credentials.

Type: string

`endpoint`

A custom endpoint URL for AWS API requests. Use this to connect to AWS-compatible services or local testing environments instead of the standard AWS endpoints.

Type: string

`global_table`

Provision the checkpoint table as a DynamoDB Global Table (v2) so checkpoints replicate across regions. Requires global_table_replicas. When the table is auto-created it is created as a global table; when it already exists, its replicas are reconciled (missing regions are added by calling UpdateTable). The existing table must have been created in global mode (TableId hash key). Enabling this against a pre-existing non-global checkpoint table fails fast with a clear error.

Requires version 4.99.0 or later.

Type: bool

Default: false

`global_table_replicas[]`

Regions other than this pipeline’s own region to replicate the checkpoint table to. The pipeline’s own region is always included. Required when global_table is true. Applied both when the checkpoint table is created and, for an existing global table, when reconciling replicas (missing regions are added; this list is not used to remove regions).

Requires version 4.99.0 or later.

Type: array<string>

Default: []

`max_tracked_shards`

Maximum number of shards holding acknowledgement state at once. A shard is normally forgotten once it has been read to the end and its last batch has been acknowledged, so this bounds the shards being read concurrently, not the shards seen over the pipeline’s lifetime. Past this limit acknowledgements return an error.

Type: int

Default: 10000

`poll_interval`

Time to wait between polling attempts when no records are available.

Type: string

Default: 1s

`region`

The AWS region in which your resources are hosted.

Type: string

`snapshot_batch_size`

Records per scan request during snapshot. Maximum 1000. Lower values provide better backpressure control but require more API calls.