org.apache.beam.sdk.io.aws2.dynamodb

Class DynamoDBIO

java.lang.Object

org.apache.beam.sdk.io.aws2.dynamodb.DynamoDBIO

@Experimental(value=SOURCE_SINK)
public final class DynamoDBIO
extends java.lang.Object

PTransforms to read/write from/to Amazon DynamoDB.

Reading from DynamoDB

Example usage:

 PCollection<List<Map<String, AttributeValue>>> output =
     pipeline.apply(
             DynamoDBIO.<List<Map<String, AttributeValue>>>read()
                 .withDynamoDbClientProvider(new BasicDynamoDbClientProvider(dynamoDbClientProvider, region))
                 .withScanRequestFn(
                     (SerializableFunction<Void, ScanRequest>)
                         input -> new ScanRequest(tableName).withTotalSegments(1))
                 .items());
 

As a client, you need to provide at least the following things:

• Specify DynamoDbClientProvider. You can pass on the default one BasicDynamoDbClientProvider
• ScanRequestFn, which you build a ScanRequest object with at least table name and total number of segment. Note This number should base on the number of your workers

PTransforms to read/write from/to DynamoDB.

Writing to DynamoDB

Example usage:

 PCollection<T> data = ...;
 data.apply(
           DynamoDBIO.<WriteRequest>write()
               .withWriteRequestMapperFn(
                   (SerializableFunction<T, KV<String, WriteRequest>>)
                       //Transforming your T data into KV<String, WriteRequest>
                       t -> KV.of(tableName, writeRequest))
               .withRetryConfiguration(
                     DynamoDBIO.RetryConfiguration.builder()
                         .setMaxAttempts(5)
                         .setMaxDuration(Duration.standardMinutes(1))
                         .build())
               .withDynamoDbClientProvider(new BasicDynamoDbClientProvider(dynamoDbClientProvider, region));
 

As a client, you need to provide at least the following things:

• Retry configuration
• Specify DynamoDbClientProvider. You can pass on the default one BasicDynamoDbClientProvider
• Mapper function with a table name to map or transform your object into KV

Note: AWS does not allow writing duplicate keys within a single batch operation. If primary keys possibly repeat in your stream (i.e. an upsert stream), you may encounter a `ValidationError`. To address this you have to provide the key names corresponding to your primary key using DynamoDBIO.Write.withDeduplicateKeys(List). Based on these keys only the last observed element is kept. Nevertheless, if no deduplication keys are provided, identical elements are still deduplicated.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	DynamoDBIO.Read<T> Read data from DynamoDB and return ScanResult.
static class	DynamoDBIO.RetryConfiguration A POJO encapsulating a configuration for retry behavior when issuing requests to DynamoDB.
static class	DynamoDBIO.Write<T> Write a PCollection data into DynamoDB.

Constructor Summary

Constructors

Constructor and Description
DynamoDBIO()

Method Summary

Modifier and Type	Method and Description
static <T> DynamoDBIO.Read<T>	read()
static <T> DynamoDBIO.Write<T>	write()

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

DynamoDBIO

public DynamoDBIO()

Method Detail

read

public static <T> DynamoDBIO.Read<T> read()

write

public static <T> DynamoDBIO.Write<T> write()