Skip to main content

aws_dynamodb_cdc

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

# Common config fields, showing default values
input:
  label: ""
  aws_dynamodb_cdc:
    tables: []
    checkpoint_table: "redpanda_dynamodb_checkpoints"
    checkpoint_namespace: ""
    start_from: "trim_horizon"
    snapshot_mode: "none"

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 a single table 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 (e.g. key:value)
  • includelist - Stream from an explicit list of tables specified in the tables field

When using tag or includelist mode, the connector will stream from all matching tables simultaneously. Each table maintains its own checkpoint state. Use table_discovery_interval to periodically rescan for new tables (useful for dynamically tagged tables).

Prerequisites​

The source DynamoDB table(s) must have 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 will first scan 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

WARNING: 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.

NOTE: 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.

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. Note that namespaces do not coordinate consumers — two pipelines sharing the same namespace will still overwrite each other's 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.

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.

Examples​

Consume CDC events​

Read change events from a DynamoDB table with streams enabled.

input:
  aws_dynamodb_cdc:
    tables: [my-table]
    region: us-east-1

Start from latest​

Only process new changes, ignoring existing stream data.

input:
  aws_dynamodb_cdc:
    tables: [orders]
    start_from: latest
    region: us-west-2

Snapshot and CDC​

Scan all existing records, then stream ongoing changes.

input:
  aws_dynamodb_cdc:
    tables: [products]
    snapshot_mode: snapshot_and_cdc
    snapshot_segments: 5
    region: us-east-1

Auto-discover tables by tag​

Automatically discover and stream from all tables with a specific tag.

input:
  aws_dynamodb_cdc:
    table_discovery_mode: tag
    table_tag_filter: "stream-enabled:true"
    table_discovery_interval: 5m
    region: us-east-1

Auto-discover tables by multiple tags​

Discover tables matching multiple tag criteria with OR logic per key, AND logic across keys.

input:
  aws_dynamodb_cdc:
    table_discovery_mode: tag
    table_tag_filter: "environment:prod,staging;team:data,analytics"
    table_discovery_interval: 5m
    region: us-east-1
    # Matches tables with: (environment=prod OR environment=staging) AND (team=data OR team=analytics)

Stream from multiple specific tables​

Stream from an explicit list of tables simultaneously.

input:
  aws_dynamodb_cdc:
    table_discovery_mode: includelist
    tables:
      - orders
      - customers
      - products
    region: us-west-2

Fields​

tables​

List of table names to stream from. For single table mode, provide one table. For multi-table mode, provide multiple tables.

Type: array of string
Default: []

table_discovery_mode​

Table discovery mode. single: stream from tables specified in tables list. tag: auto-discover tables by tags (ignores tables field). includelist: stream from tables in tables list (alias for single, kept for compatibility).

Type: string
Default: "single"

Options: single, tag, includelist

table_tag_filter​

Multi-tag filter: 'key1:v1,v2;key2:v3,v4'. Matches tables with (key1=v1 OR key1=v2) AND (key2=v3 OR key2=v4). Required when table_discovery_mode is tag.

Type: string
Default: ""

table_discovery_interval​

Interval for rescanning and discovering new tables when using tag or includelist mode. Set to 0 to disable periodic rescanning.

Type: string
Default: "5m"

checkpoint_table​

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

Type: string
Default: "redpanda_dynamodb_checkpoints"

checkpoint_namespace​

An optional namespace for checkpoints, allowing multiple independent pipelines (for example one per developer or environment) to share a single checkpoint table without overwriting each other's positions. Checkpoints written under one namespace are invisible to pipelines using a different namespace (or none), so changing this value causes the pipeline to restart from start_from. Must not contain #.

Type: string
Default: ""

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 via 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.

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).

Type: array of string
Default: []

batch_size​

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

Type: int
Default: 1000

poll_interval​

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

Type: string
Default: "1s"

start_from​

Where to start reading when no checkpoint exists. trim_horizon starts from the oldest available record, latest starts from new records.

Type: string
Default: "trim_horizon"

Options: trim_horizon, latest

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

max_tracked_shards​

Maximum number of shards to track simultaneously. Prevents memory issues with extremely large tables.

Type: int
Default: 10000

throttle_backoff​

Time to wait when applying backpressure due to too many in-flight messages.

Type: string
Default: "100ms"

snapshot_mode​

Snapshot behavior. none: CDC only (default). snapshot_only: one-time table scan, no streaming. snapshot_and_cdc: scan entire table then stream changes.

Type: string
Default: "none"

Options: none, snapshot_only, snapshot_and_cdc

snapshot_segments​

Number of parallel scan segments (1-10). Higher parallelism scans faster but consumes more RCUs. Start with 1 for safety.

Type: int
Default: 1

snapshot_batch_size​

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

Type: int
Default: 100

snapshot_throttle​

Minimum time between scan requests per segment. Use this to limit RCU consumption during snapshot.

Type: string
Default: "100ms"

snapshot_deduplicate​

Deduplicate records that appear in both snapshot and CDC stream. Requires buffering CDC events during snapshot. If buffer is exceeded, deduplication is disabled to prevent data loss.

Type: bool
Default: true

snapshot_buffer_size​

Maximum CDC events to buffer for deduplication (approximately 100 bytes per entry). If exceeded, deduplication is disabled and duplicates may be emitted.

Type: int
Default: 100000

region​

The AWS region to target.

Type: string

endpoint​

Allows you to specify a custom endpoint for the AWS API.

Type: string

tcp​

TCP socket configuration.

Type: object

tcp.connect_timeout​

Maximum amount of time a dial will wait for a connect to complete. Zero disables.

Type: string
Default: "0s"

tcp.keep_alive​

TCP keep-alive probe configuration.

Type: object

tcp.keep_alive.idle​

Duration the connection must be idle before sending the first keep-alive probe. Zero defaults to 15s. Negative values disable keep-alive probes.

Type: string
Default: "15s"

tcp.keep_alive.interval​

Duration between keep-alive probes. Zero defaults to 15s.

Type: string
Default: "15s"

tcp.keep_alive.count​

Maximum unanswered keep-alive probes before dropping the connection. Zero defaults to 9.

Type: int
Default: 9

tcp.tcp_user_timeout​

Maximum time to wait for acknowledgment of transmitted data before killing the connection. Linux-only (kernel 2.6.37+), ignored on other platforms. When enabled, keep_alive.idle must be greater than this value per RFC 5482. Zero disables.

Type: string
Default: "0s"

credentials​

Optional manual configuration of AWS credentials to use. More information can be found in this document.

Type: object

credentials.profile​

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

Type: string

credentials.id​

The ID of credentials to use.

Type: string

credentials.secret​

The secret for the credentials being used.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string

credentials.token​

The token for the credentials being used, required when using short term credentials.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string

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.role​

A role ARN to assume.

Type: string

credentials.role_external_id​

An external ID to provide when assuming a role.

Type: string