AWS DynamoDB Streams component

Q: Why do I get no records?

The table needs a stream (StreamSpecification), and with the default FROM_LATEST only changes made after the consumer started are delivered. Use FROM_START to read the retained records.

Q: Which images are in the record?

That depends on the table's stream view type: KEYS_ONLY, NEW_IMAGE, OLD_IMAGE or NEW_AND_OLD_IMAGES. You can change it with aws2-ddb UpdateTable and the CamelAwsDdbStreamViewType header.

The aws2-ddbstream component consumes the change records (inserts, updates, deletes) of an Amazon DynamoDB table's stream.

Scheme: aws2-ddbstream: · Crate: cameleer-cloud · Cargo feature: cloud · Consumer: yes · Producer: no · Apache Camel: AWS DynamoDB Streams

When to use it

Change data capture: react to every new or changed item without polling the table.
Replicate a table into a search index, a cache or another region.
Trigger notifications or workflows when an item is inserted.

URI format

text

aws2-ddbstream://tableName[?options]

The stream must be enabled on the table. Its ARN is read with DescribeTable.

Example

Forward every new order item to a queue:

yaml example: aws2-ddbstream-new-orders

- route:
    id: new-orders-cdc
    from:
      uri: "aws2-ddbstream://orders?region={{env:AWS_REGION:us-east-1}}&streamIteratorType=FROM_START"
      steps:
        - filter:
            simple: "${header.CamelAwsDdbStreamEventName} == 'INSERT'"
            steps:
              - marshal:
                  json: {}
              - to: "aws2-sqs://new-orders?region={{env:AWS_REGION:us-east-1}}&autoCreateQueue=true"

The body is the stream record as a map: eventID, eventName (INSERT, MODIFY, REMOVE), eventSource, awsRegion and dynamodb with Keys, NewImage and OldImage already converted to plain maps, plus SequenceNumber, SizeBytes and StreamViewType.

Rust DSL

rust

ctx.add_routes(|b| {
    b.from("aws2-ddbstream://orders?region={{env:AWS_REGION:us-east-1}}&streamIteratorType=FROM_START")
        .filter(simple("${header.CamelAwsDdbStreamEventName} == 'INSERT'"))
        .marshal("json")
        .to("aws2-sqs://new-orders?region={{env:AWS_REGION:us-east-1}}&autoCreateQueue=true");
})?;

Options

Common AWS options (region, credentials, endpoint override, proxy) are listed in Cloud credentials.

option	default	description
`streamIteratorType`	`FROM_LATEST`	`FROM_LATEST`, `FROM_START` or `FROM_SEQUENCE_NUMBER`.
`sequenceNumberProvider`		Start sequence number for `FROM_SEQUENCE_NUMBER`.
`maxResultsPerRequest`	`100`	Records per `GetRecords` call.

The consumer also accepts the scheduled poll options (delay default 500 ms, ...). New child shards after a split are picked up and read from their start; expired iterators are re-created after the last sequence number seen.

Message headers

header	direction	description
`CamelAwsDdbStreamEventName`	out	Cameleer: `INSERT`, `MODIFY` or `REMOVE`.
`CamelAwsDdbStreamSequenceNumber`	out	Cameleer: sequence number of the record.
`CamelAwsDdbStreamShardId`	out	Cameleer: shard of the record.

Run it with Docker

docker run --rm -e AWS_REGION -e AWS_ACCESS_KEY_ID -e AWS_SECRET_ACCESS_KEY \
  -v "$PWD:/routes:ro" pmsudhi/cameleer run /routes/cdc.camel.yaml

Common questions

Why do I get no records?

The table needs a stream (StreamSpecification), and with the default FROM_LATEST only changes made after the consumer started are delivered. Use FROM_START to read the retained records.

Which images are in the record?

That depends on the table's stream view type: KEYS_ONLY, NEW_IMAGE, OLD_IMAGE or NEW_AND_OLD_IMAGES. You can change it with aws2-ddb UpdateTable and the CamelAwsDdbStreamViewType header.

Found a problem?

If something on this page doesn't match what Cameleer does, or you hit a bug while building with it, please help fix it, whether you are a person or a coding agent: open an issue that describes what you saw, then fork the repository, fix it, add or update a test, run cargo test --workspace, and open a pull request. CONTRIBUTING.md explains the ground rules.