dynamodb

The dynamodb source enables change data capture (CDC) on Amazon DynamoDB tables. It can receive table events, such as create, update, or delete, using DynamoDB streams and supports initial snapshots using point-in-time recovery (PITR).

The source includes two ingestion options to stream DynamoDB events:

1. A full initial snapshot using PITR gets an initial snapshot of the current state of the DynamoDB table. This requires the PITR Snapshots and DyanmoDB option enabled on your DynamoDB table.
2. Stream events from DynamoDB streams without full initial snapshots. This is useful if you already have a snapshot mechanism within your pipelines. This requires that the DynamoDB stream option is enabled on the DynamoDB table.

Usage

The following example pipeline specifies DynamoDB as a source. It ingests data from a DyanmoDB table named table-a through a PITR snapshot. It also indicates the start_position, which tells the pipeline how to read DynamoDB stream events:

version: "2"
cdc-pipeline:
  source:
    dynamodb:
      tables:
        - table_arn: "arn:aws:dynamodb:us-west-2:123456789012:table/table-a"
          export:
            s3_bucket: "test-bucket"
            s3_prefix: "myprefix"
          stream:
            start_position: "LATEST" # Read latest data from streams (Default)
            view_on_remove: NEW_IMAGE
      aws:
        region: "us-west-2"
        sts_role_arn: "arn:aws:iam::123456789012:role/my-iam-role"

Configuration options

The following tables describe the configuration options for the dynamodb source.

aws

Use the following options in the AWS configuration.

tables

Use the following options with the tables configuration.

Export options

The following options let you customize the export destination for DynamoDB events.

Stream option

The following option lets you customize how the pipeline reads events from the DynamoDB table.

Exposed metadata attributes

The following metadata will be added to each event that is processed by the dynamodb source. These metadata attributes can be accessed using the expression syntax getMetadata function.

• primary_key: The primary key of the DynamoDB item. For tables that only contain a partition key, this value provides the partition key. For tables that contain both a partition and sort key, the primary_key attribute will be equal to the partition and sort key, separated by a |, for example, partition_key|sort_key.
• partition_key: The partition key of the DynamoDB item.
• sort_key: The sort key of the DynamoDB item. This will be null if the table does not contain a sort key.
• dynamodb_timestamp: The timestamp of the DynamoDB item. This will be the export time for export items and the DynamoDB stream event time for stream items. This timestamp is used by sinks to emit an EndtoEndLatency metric for DynamoDB stream events that tracks the latency between a change occurring in the DynamoDB table and that change being applied to the sink.
• document_version: Uses the dynamodb_timestamp to modify break ties between stream items that are received in the same second. Recommend for use with the opensearch sink’s document_version setting.
• opensearch_action: A default value for mapping DynamoDB event actions to OpenSearch actions. This action will be index for export items, and INSERT or MODIFY for stream events, and REMOVE stream events when the OpenSearch action is delete.
• dynamodb_event_name: The exact event type for the item. Will be null for export items and either INSERT, MODIFY, or REMOVE for stream events.
• table_name: The name of the DynamoDB table that an event came from.

Permissions

The following are the minimum required permissions for running DynamoDB as a source:

{
    "Version": "2012-10-17",
    "Statement": [
      {
        "Sid": "allowDescribeTable",
        "Effect": "Allow",
        "Action": [
          "dynamodb:DescribeTable"
        ],
        "Resource": [
          "arn:aws:dynamodb:us-east-1:{account-id}:table/my-table"
        ]
      },
       {
            "Sid": "allowRunExportJob",
            "Effect": "Allow",
            "Action": [
                "dynamodb:DescribeContinuousBackups",
                "dynamodb:ExportTableToPointInTime"
            ],
            "Resource": [
                "arn:aws:dynamodb:us-east-1:{account-id}:table/my-table"
            ]
        },
        {
            "Sid": "allowCheckExportjob",
            "Effect": "Allow",
            "Action": [
                "dynamodb:DescribeExport"
            ],
            "Resource": [
                "arn:aws:dynamodb:us-east-1:{account-id}:table/my-table/export/*"
            ]
        },
        {
            "Sid": "allowReadFromStream",
            "Effect": "Allow",
            "Action": [
                "dynamodb:DescribeStream",
                "dynamodb:GetRecords",
                "dynamodb:GetShardIterator"
            ],
            "Resource": [
                "arn:aws:dynamodb:us-east-1:{account-id}:table/my-table/stream/*"
            ]
        },
        {
            "Sid": "allowReadAndWriteToS3ForExport",
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:AbortMultipartUpload",
                "s3:PutObject",
                "s3:PutObjectAcl"
            ],
            "Resource": [
                "arn:aws:s3:::my-bucket/*"
            ]
        }
    ]
}

When performing an export, the "Sid": "allowReadFromStream" section is not required. If only reading from DynamoDB streams, the "Sid": "allowReadAndWriteToS3ForExport", "Sid": "allowCheckExportjob", and ` “Sid”: “allowRunExportJob”` sections are not required.

Metrics

The dynamodb source includes the following metrics.

Counters

• exportJobSuccess: The number of export jobs that have been submitted successfully.
• exportJobFailure: The number of export job submission attempts that have failed.
• exportS3ObjectsTotal: The total number of export data files found in S3.
• exportS3ObjectsProcessed: The total number of export data files that have been processed successfully from S3.
• exportRecordsTotal: The total number of records found in the export.
• exportRecordsProcessed: The total number of export records that have been processed successfully.
• exportRecordsProcessingErrors: The number of export record processing errors.
• changeEventsProcessed: The number of change events processed from DynamoDB streams.
• changeEventsProcessingErrors: The number of processing errors for change events from DynamoDB streams.
• shardProgress: The incremented shard progress when DynamoDB streams are being read correctly. This being0 for any significant amount of time means there is a problem with the pipeline that has streams enabled.