Skip to main content

aws_dynamodb_cdc

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

Stream item-level changes from DynamoDB tables using DynamoDB Streams. This input automatically manages shards, checkpoints progress for recovery, and processes multiple shards concurrently.

# Config fields, showing default values
input:
  label: ""
  aws_dynamodb_cdc:
    tables: []
    table_discovery_mode: single
    table_discovery_interval: 5m
    table_tag_filter: ""
    checkpoint_table: redpanda_dynamodb_checkpoints
    checkpoint_limit: 1000
    batch_size: 1000
    poll_interval: 1s
    start_from: trim_horizon
    max_tracked_shards: 10000
    throttle_backoff: 100ms
    snapshot_mode: none
    snapshot_segments: 1
    snapshot_batch_size: 100
    snapshot_buffer_size: 100000
    snapshot_throttle: 100ms
    snapshot_deduplicate: true
    region: ""
    endpoint: ""
    credentials:
      profile: ""
      id: ""
      secret: ""
      token: ""
      from_ec2_role: false
      role: ""
      role_external_id: ""
    tcp:
      connect_timeout: ""
      keep_alive:
        count: 0
        idle: ""
        interval: ""
      tcp_user_timeout: ""

Examples​

Stream change events from a single named table.

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

Fields​

tables​

A list of DynamoDB table names to stream change events from.

Type: array
Default: []

table_discovery_mode​

How the tables to stream from are selected. single streams from the named tables, tag discovers tables that match table_tag_filter, and includelist streams from the explicit list provided in tables.

Type: string
Default: "single"

table_discovery_interval​

The interval at which to rescan for and discover new tables.

Type: string
Default: "5m"

table_tag_filter​

A multi-tag filter used to discover tables when table_discovery_mode is tag, in the format key1:v1,v2;key2:v3,v4.

Type: string
Default: ""

checkpoint_table​

The DynamoDB table name used to store checkpoints. It will be created if it does not exist.

Type: string
Default: "redpanda_dynamodb_checkpoints"

checkpoint_limit​

The maximum number of unacknowledged messages before forcing a checkpoint update.

Type: int
Default: 1000

batch_size​

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

Type: int
Default: 1000

poll_interval​

The time to wait between polling attempts when no records are available.

Type: string
Default: "1s"

start_from​

Where to begin reading a shard when no checkpoint exists. trim_horizon reads from the oldest available record and latest reads only new records.

Type: string
Default: "trim_horizon"

max_tracked_shards​

The maximum number of shards to track simultaneously.

Type: int
Default: 10000

throttle_backoff​

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

Type: string
Default: "100ms"

snapshot_mode​

Controls whether an initial table snapshot is taken before streaming CDC events. none streams CDC events only, snapshot_only reads the current table contents once, and snapshot_and_cdc takes a snapshot and then continues streaming changes.

Type: string
Default: "none"

snapshot_segments​

The number of parallel scan segments to use during a snapshot. Valid range: 1 to 10.

Type: int
Default: 1

snapshot_batch_size​

The number of records to read per scan request during a snapshot. Maximum 1000.

Type: int
Default: 100

snapshot_buffer_size​

The maximum number of CDC events to buffer for deduplication (roughly 100 bytes per entry).

Type: int
Default: 100000

snapshot_throttle​

The minimum time between scan requests per segment during a snapshot.

Type: string
Default: "100ms"

snapshot_deduplicate​

Whether to deduplicate records that appear in both the snapshot and the CDC stream.

Type: bool
Default: true

region​

The AWS region to target.

Type: string
Default: ""

endpoint​

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

Type: string
Default: ""

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
Default: ""

credentials.id​

The ID of credentials to use.

Type: string
Default: ""

credentials.secret​

The secret for the credentials being used. :::warning Secret This field contains sensitive information that usually shouldn't be added to a config directly, read our secrets page for more info. :::

Type: string
Default: ""

credentials.token​

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

Type: string
Default: ""

credentials.from_ec2_role​

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

Type: bool
Default: false

credentials.role​

A role ARN to assume.

Type: string
Default: ""

credentials.role_external_id​

An external ID to provide when assuming a role.

Type: string
Default: ""

tcp​

Advanced options for the underlying TCP connection to the AWS API, including connection timeout and keep-alive settings.

Type: object

tcp.connect_timeout​

The maximum time to wait when establishing the TCP connection to the AWS API.

Type: string
Default: ""

tcp.keep_alive​

TCP keep-alive settings for the connection.

Type: object

tcp.keep_alive.count​

The maximum number of keep-alive probes to send before dropping the connection.

Type: int
Default: 0

tcp.keep_alive.idle​

The time a connection can remain idle before keep-alive probes are sent.

Type: string
Default: ""

tcp.keep_alive.interval​

The interval between keep-alive probes.

Type: string
Default: ""

tcp.tcp_user_timeout​

The maximum time transmitted data may remain unacknowledged before the connection is closed.

Type: string
Default: ""