Amazon DynamoDB SQL Connector

Sink: Batch Sink: Streaming Append & Upsert Mode

The DynamoDB connector allows for writing data into Amazon DynamoDB.

Dependencies

There is no connector (yet) available for Flink version 1.20.

How to create a DynamoDB table

Follow the instructions from the Amazon DynamoDB Developer Guide to set up a DynamoDB table. The following example shows how to create a table backed by a DynamoDB table with minimum required options:

CREATE TABLE DynamoDbTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `category_id` BIGINT,
  `behavior` STRING
)
WITH (
  'connector' = 'dynamodb',
  'table-name' = 'user_behavior',
  'aws.region' = 'us-east-2'
);

Connector Options

Authorization

Make sure to create an appropriate IAM policy to allow writing to the DynamoDB table.

Authentication

Depending on your deployment you would choose an appropriate Credentials Provider to allow access to DynamoDB. By default, the AUTO Credentials Provider is used. If the access key ID and secret key are set in the deployment configuration, this results in using the BASIC provider.

A specific AWSCredentialsProvider can be optionally set using the aws.credentials.provider setting. Supported values are:

• AUTO - Use the default AWS Credentials Provider chain that searches for credentials in the following order: ENV_VARS, SYS_PROPS, WEB_IDENTITY_TOKEN, PROFILE, and EC2/ECS credentials provider.
• BASIC - Use access key ID and secret key supplied as configuration.
• ENV_VAR - Use AWS_ACCESS_KEY_ID & AWS_SECRET_ACCESS_KEY environment variables.
• SYS_PROP - Use Java system properties aws.accessKeyId and aws.secretKey.
• PROFILE - Use an AWS credentials profile to create the AWS credentials.
• ASSUME_ROLE - Create AWS credentials by assuming a role. The credentials for assuming the role must be supplied.
• WEB_IDENTITY_TOKEN - Create AWS credentials by assuming a role using Web Identity Token.
• CUSTOM - Provide a custom class that implements the interface AWSCredentialsProvider and has a constructor MyCustomClass(java.util.Properties config). All connector properties will be passed down to this custom credential provider class via the constructor.

Sink Partitioning

The DynamoDB sink supports client side deduplication of data via the PARTITIONED BY clause. You can specify a list of partition keys, the sink will only send the latest record for each composite key within a batch. For example:

CREATE TABLE DynamoDbTable (
  `user_id` BIGINT,
  `item_id` BIGINT,
  `category_id` BIGINT,
  `behavior` STRING
) PARTITIONED BY ( user_id )
WITH (
  'connector' = 'dynamodb',
  'table-name' = 'user_behavior',
  'aws.region' = 'us-east-2'
);

Notice

The current implementation of the DynamoDB SQL connector is write-only and doesn’t provide an implementation for source queries. Queries similar to:

SELECT * FROM DynamoDbTable;

should result in an error similar to

Connector dynamodb can only be used as a sink. It cannot be used as a source.