Version: Next

AmazonDynamoDB

Amazon DynamoDB source connector

Description

The Amazon DynamoDB source connector reads existing items from an Amazon DynamoDB table by using DynamoDB scan requests.

The connector is a batch source. DynamoDB does not expose field types in the same way as a relational database, so the SeaTunnel schema must be configured explicitly.

This source reads the current table data with scan requests. It does not read DynamoDB Streams or CDC change events.

Use tables_configs to read several DynamoDB tables with different schemas through one source.

Supported Engines

Spark
Flink
SeaTunnel Zeta

Key Features

Options

name	type	required	default value	description
url	string	yes	-	DynamoDB endpoint URL.
region	string	yes	-	AWS region of the DynamoDB service.
access_key_id	string	yes	-	AWS access key ID.
secret_access_key	string	yes	-	AWS secret access key.
table	string	no	-	Required in single-table mode; mutually exclusive with `tables_configs`.
schema	config	no	-	Required in single-table mode; define it per entry in multi-table mode.
tables_configs	list	no	-	Tables and schemas to read in multi-table mode; see below.
scan_item_limit	int	no	1	Maximum items returned by each scan request.
parallel_scan_threads	int	no	2	Number of logical segments for parallel scan.
common-options	object	no	-	Source plugin common parameters.

url [string]

The DynamoDB endpoint URL, for example https://dynamodb.us-east-1.amazonaws.com.

When testing with DynamoDB Local, use the local endpoint, for example http://127.0.0.1:8000.

region [string]

The AWS region of the DynamoDB service, such as us-east-1.

access_key_id [string]

The AWS access key ID used to connect to DynamoDB.

secret_access_key [string]

The AWS secret access key used to connect to DynamoDB.

table [string]

The DynamoDB table name to scan in single-table mode.

schema [config]

Defines the SeaTunnel fields to read from DynamoDB items.

DynamoDB is a key-value and document database. The source connector cannot infer a complete SeaTunnel schema from DynamoDB, so every field that should be read must be listed here.

schema = {
  fields {
    id = string
    c_map = "map<string, smallint>"
    c_array = "array<tinyint>"
    c_string = string
    c_boolean = boolean
    c_int = int
    c_bigint = bigint
    c_float = float
    c_double = double
    c_decimal = "decimal(2, 1)"
    c_bytes = bytes
    c_date = date
    c_timestamp = timestamp
  }
}

For more schema syntax, see Schema Feature.

tables_configs [list]

An alternative to the root-level table and schema for reading several tables. Each entry requires:

table: the DynamoDB table name to scan.
schema: the SeaTunnel fields to read from that table. schema.table sets the output table identity; when it is not set, the DynamoDB table name is used. A dot in the name is read as a database.table separator, so set schema.table when the DynamoDB table name contains dots.

An entry can also set scan_item_limit and parallel_scan_threads. When an entry does not set them, the root-level values (or their defaults) are used. Each table is scanned with its own parallel_scan_threads segments, and each row carries the identity of the table it was read from.

All entries share the root connection options url, region, access_key_id and secret_access_key; connection options inside an entry are rejected. A root-level table or schema, an empty list, an entry without table or schema, unsupported entry options and duplicate output table identities are rejected before the job starts. Keep the table identities stable when restoring a checkpoint; switching between single-table and multi-table mode requires a fresh job.

scan_item_limit [int]

The maximum number of items returned by each DynamoDB scan request.

Larger values reduce the number of requests but may increase the memory used by each read batch.

parallel_scan_threads [int]

The number of logical scan segments used for DynamoDB parallel scan.

This value controls how the source splits the table scan. It should usually be aligned with job parallelism and table size.

For small tables, keep the default value. For large tables, increase it together with env.parallelism and the source parallelism option so that multiple readers can scan different segments.

common options

Source plugin common parameters, please refer to Source Common Options for details.

Usage Notes

The source uses DynamoDB scan requests, so it reads the current table snapshot rather than change events.
access_key_id and secret_access_key are required by this connector. For DynamoDB Local, use dummy values accepted by the local service.
parallel_scan_threads controls the number of DynamoDB scan segments. Increase it together with job parallelism for larger tables.
scan_item_limit is the page limit used on each scan request, not the total number of rows in the job.

Data Type Mapping

SeaTunnel Data Type	DynamoDB Attribute Type
BOOLEAN	BOOL
TINYINT	N
SMALLINT	N
INT	N
BIGINT	N
FLOAT	N
DOUBLE	N
DECIMAL	N
STRING	S
TIME	S
DATE	S
TIMESTAMP	S
BYTES	B
MAP	M
ARRAY	L
NULL	NULL

Task Example

Read One Table

The following example reads rows from source_table and writes them to sink_table.

env {
  parallelism = 2
  job.mode = "BATCH"
}

source {
  AmazonDynamoDB {
    url = "http://127.0.0.1:8000"
    region = "us-east-1"
    access_key_id = "dummy-key"
    secret_access_key = "dummy-secret"
    table = "source_table"
    parallelism = 2
    scan_item_limit = 2
    parallel_scan_threads = 4
    schema = {
      fields {
        id = string
        c_map = "map<string, smallint>"
        c_array = "array<tinyint>"
        c_string = string
        c_boolean = boolean
        c_tinyint = tinyint
        c_smallint = smallint
        c_int = int
        c_bigint = bigint
        c_float = float
        c_double = double
        c_decimal = "decimal(2, 1)"
        c_bytes = bytes
        c_date = date
        c_timestamp = timestamp
      }
    }
  }
}

sink {
  AmazonDynamoDB {
    url = "http://127.0.0.1:8000"
    region = "us-east-1"
    access_key_id = "dummy-key"
    secret_access_key = "dummy-secret"
    table = "sink_table"
    batch_size = 25
  }
}

Read Multiple Tables

The following example reads orders and customers with different schemas. Rows from customers are routed downstream as crm.customers.

env {
  parallelism = 2
  job.mode = "BATCH"
}

source {
  AmazonDynamoDB {
    url = "http://127.0.0.1:8000"
    region = "us-east-1"
    access_key_id = "dummy-key"
    secret_access_key = "dummy-secret"
    parallel_scan_threads = 2
    tables_configs = [
      {
        table = "orders"
        parallel_scan_threads = 4
        schema = {
          fields {
            id = string
            amount = int
          }
        }
      },
      {
        table = "customers"
        schema = {
          table = "crm.customers"
          fields {
            id = string
            name = string
            vip = boolean
          }
        }
      }
    ]
  }
}

sink {
  Console {}
}

Changelog

Change Log

Change	Commit	Version
[Improve][Core] Unify the aws-sdk-v2 version to 2.31.30 (#9698)	https://github.com/apache/seatunnel/commit/41c251cc8a	2.3.12
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)	https://github.com/apache/seatunnel/commit/4f5adeb1c7	2.3.11
[Fix] Fix error log name for SourceSplitEnumerator implements class (#8817)	https://github.com/apache/seatunnel/commit/55ed90ecaf	2.3.10
[Improve] restruct connector common options (#8634)	https://github.com/apache/seatunnel/commit/f3499a6eeb	2.3.10
[improve] update amazondynamodb connector (#8601)	https://github.com/apache/seatunnel/commit/a69efca0fd	2.3.10
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)	https://github.com/apache/seatunnel/commit/6b7c53d03c	2.3.9
[Feature][Doris] Add Doris type converter (#6354)	https://github.com/apache/seatunnel/commit/5189991843	2.3.6
[Feature][Core] Upgrade flink source translation (#5100)	https://github.com/apache/seatunnel/commit/5aabb14a94	2.3.4
[Improve][Common] Introduce new error define rule (#5793)	https://github.com/apache/seatunnel/commit/9d1b2582b2	2.3.4
[Improve] Code clean for AmazonDynamoDB connector (#5791)	https://github.com/apache/seatunnel/commit/a17dd7afc1	2.3.4
[Improve] Remove use `SeaTunnelSink::getConsumedType` method and mark it as deprecated (#5755)	https://github.com/apache/seatunnel/commit/8de7408100	2.3.4
[connector-v2] add amazondynamicdb source split (#5275)	https://github.com/apache/seatunnel/commit/740c14422d	2.3.4
Support config column/primaryKey/constraintKey in schema (#5564)	https://github.com/apache/seatunnel/commit/eac76b4e50	2.3.4
[Improve]Remove scheduler in Dynamodb sink (#5248)	https://github.com/apache/seatunnel/commit/9e033a824e	2.3.4
[Improve][CheckStyle] Remove useless 'SuppressWarnings' annotation of checkstyle. (#5260)	https://github.com/apache/seatunnel/commit/51c0d709ba	2.3.4
[Bugfix][AmazonDynamoDB] Fix the problem that all table data cannot be obtained (#5146)	https://github.com/apache/seatunnel/commit/09995159a0	2.3.3
Merge branch 'dev' into merge/cdc	https://github.com/apache/seatunnel/commit/4324ee1912	2.3.1
[Improve][Project] Code format with spotless plugin.	https://github.com/apache/seatunnel/commit/423b583038	2.3.1
[improve][api] Refactoring schema parse (#4157)	https://github.com/apache/seatunnel/commit/b2f573a13e	2.3.1
[Improve][build] Give the maven module a human readable name (#4114)	https://github.com/apache/seatunnel/commit/d7cd601051	2.3.1
[Improve][Project] Code format with spotless plugin. (#4101)	https://github.com/apache/seatunnel/commit/a2ab166561	2.3.1
[Feature][Connector] add get source method to all source connector (#3846)	https://github.com/apache/seatunnel/commit/417178fb84	2.3.1
[Feature][API & Connector & Doc] add parallelism and column projection interface (#3829)	https://github.com/apache/seatunnel/commit/b9164b8ba1	2.3.1
[Hotfix][API] env required option can not set default value (#3584)	https://github.com/apache/seatunnel/commit/c5a23024f6	2.3.0
[Feature][Connector-V2][AmazonDynamoDB] Add Factory for AmazonDynamoDB (#3348)	https://github.com/apache/seatunnel/commit/a0068efdbf	2.3.0
[Improve][Connector-V2][AmazonDynamoDB] Unified exception for AmazonDynamoDB source & sink connector (#3333)	https://github.com/apache/seatunnel/commit/17bc5adcef	2.3.0
[Connector-V2][Chore] Canonical name for AmazonDynamodb (#3321)	https://github.com/apache/seatunnel/commit/e216eb9a6b	2.3.0
[Feature][Connector-V2] [Amazondynamodb Connector]add amazondynamodb source & sink connnector (#3166)	https://github.com/apache/seatunnel/commit/183bac02f0	2.3.0

AmazonDynamoDB

Description​

Supported Engines​

Key Features​

Options​

url [string]​

region [string]​

access_key_id [string]​

secret_access_key [string]​

table [string]​

schema [config]​

tables_configs [list]​

scan_item_limit [int]​

parallel_scan_threads [int]​

common options​

Usage Notes​

Data Type Mapping​

Task Example​

Read One Table​

Read Multiple Tables​

Changelog​