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

If primary keys could repeat in your stream (i.e. an upsert stream), you could encounter a ValidationError, as AWS does not allow writing duplicate keys within a single batch operation. For such use cases, you can explicitly set the key names corresponding to the primary key to be deduplicated using the withDeduplicateKeys method

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()