DynamoDB

The DynamoDB source copies a table with a parallel scan and then follows its DynamoDB Stream, shard by shard, parents before children.

Prerequisites

The table needs a stream whose view type includes new images: NEW_AND_OLD_IMAGES (recommended) or NEW_IMAGE.

aws dynamodb update-table --table-name products \
  --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES

Without a stream, or with KEYS_ONLY / OLD_IMAGE, the task fails and prints the command above. Changing the view type means disabling the stream and enabling it again, which creates a new stream: resync the task afterwards.

IAM permissions

A policy for a dedicated IAM user or role, scoped to the table and its streams:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["dynamodb:DescribeTable", "dynamodb:Scan"],
      "Resource": "arn:aws:dynamodb:eu-west-1:123456789012:table/products"
    },
    {
      "Effect": "Allow",
      "Action": ["dynamodb:DescribeStream", "dynamodb:GetShardIterator", "dynamodb:GetRecords"],
      "Resource": "arn:aws:dynamodb:eu-west-1:123456789012:table/products/stream/*"
    },
    {
      "Effect": "Allow",
      "Action": ["dynamodb:ListTables"],
      "Resource": "*"
    }
  ]
}

dynamodb:ListTables is only used by the connector's table browser and test.

Connection settings

The task only adds table, the DynamoDB table name.

How changes are applied

• Primary key. The table's partition key, followed by its sort key when it has one. A table with both gets a composite document id.
• Schema. Attributes are inferred from a scan of 100 items; attributes with several types show as mixed.
• Snapshot. A parallel scan with 4 segments.
• Inserts and modifications are written from the new image. Removals (including TTL expirations) delete the document by its keys.
• Values. Numbers without a fraction become integers, sets and lists become arrays, maps become objects, binary becomes base64.
• DynamoDB has no truncate; deleting the table ends its stream.

Positions and resume

A position is a JSON checkpoint holding, for every shard being read, the sequence number of the last record delivered, plus the shards already read to their end and the time the position was first taken. Shards are polled every second and the shard list is refreshed every 10 seconds.

Stream records are kept for 24 hours. If a task stays stopped longer, or a shard expires before it was fully read (TrimmedDataAccessException), the position is lost and the task resyncs automatically.

Troubleshooting