org.apache.beam.sdk.io.aws.dynamodb

Class DynamoDBIO

java.lang.Object

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

Deprecated.

Module beam-sdks-java-io-amazon-web-services is deprecated and will be eventually removed. Please migrate to DynamoDBIO in module beam-sdks-java-io-amazon-web-services2.

@Deprecated
public final class DynamoDBIO
extends java.lang.Object

PTransforms to read/write from/to Amazon 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.create(5, Duration.standardMinutes(1)))
               .withAwsClientsProvider(new BasicDynamoDbProvider(accessKey, secretKey, region));
 

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

• Retry configuration
• Specify AwsClientsProvider. You can pass on the default one BasicDynamoDbProvider
• 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.

Reading from DynamoDB

Example usage:

 PCollection<List<Map<String, AttributeValue>>> output =
     pipeline.apply(
             DynamoDBIO.<List<Map<String, AttributeValue>>>read()
                 .withAwsClientsProvider(new BasicDynamoDBProvider(accessKey, secretKey, 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 AwsClientsProvider. You can pass on the default one BasicDynamoDBProvider
• 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

Nested Class Summary

Nested Classes

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

Constructor Summary

Constructors

Constructor and Description
DynamoDBIO() Deprecated.

Method Summary

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

Methods inherited from class java.lang.Object

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

Constructor Detail

DynamoDBIO

public DynamoDBIO()

Deprecated.

Method Detail

read

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

Deprecated.

write

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

Deprecated.