DynamoDB - 表活动

DynamoDB流使您可以跟踪和响应表项更改.使用此功能可以创建一个应用程序，该应用程序通过跨源更新信息来响应更改.为大型多用户系统的数千名用户同步数据.使用它可以在更新时向用户发送通知.它的应用证明是多种多样的. DynamoDB流作为用于实现此功能的主要工具.

流捕获包含表内项修改的按时间排序的序列.他们持有这些数据最多24小时.应用程序使用它们几乎实时地查看原始项目和修改项目.

表上启用的流捕获所有修改.在任何CRUD操作上，DynamoDB都会使用已修改项的主键属性创建流记录.您可以配置流以获取其他信息，例如图像之前和之后.

Streams带有两个保证 :

• 每条记录在流中出现一次

• 每项修改都会产生与该记录顺序相同的流记录修改.

所有流都是实时处理的，允许您将它们用于应用程序中的相关功能.

管理流

在创建表时，您可以启用流.现有表允许流禁用或设置更改. Streams提供异步操作的功能，这意味着不会对表性能产生影响.

利用AWS管理控制台进行简单的流管理.首先，导航到控制台，然后选择表.在"概览"标签中，选择管理流.在窗口内，选择添加到表数据修改的流中的信息.输入所有设置后，选择启用.

如果要禁用任何现有流，请选择管理流，然后选择禁用.

您还可以使用API CreateTable和UpdateTable来启用或更改流.使用参数StreamSpecification配置流. StreamEnabled指定状态，对于启用表示true，对于禁用表示false.

StreamViewType指定添加到流的信息:KEYS_ONLY，NEW_IMAGE，OLD_IMAGE和NEW_AND_OLD_IMAGES.

流读取

通过连接到端点并发出API请求来读取和处理流.每个流由流记录组成，每个记录作为拥有流的单个修改存在.流记录包括显示发布顺序的序列号.记录属于也称为分片的组.碎片用作多个记录的容器，还包含访问和遍历记录所需的信息. 24小时后，记录会自动删除.

这些碎片会根据需要生成和删除，并且不会持续很长时间.它们还会自动分成多个新的分片，通常是为了响应写入活动的峰值.在流禁用时，打开碎片关闭.分片之间的层次关系意味着应用程序必须优先考虑父分片以获得正确的处理顺序.您可以使用Kinesis Adapter自动执行此操作.

注意 : 导致无变化的操作不会写入流记录.

访问和处理记录需要执行以下任务 :

• 确定目标流的ARN.

• 确定保存目标记录的流的分片.

• 访问分片以检索所需的记录.

注意 : 一次最多应该有两个进程读取一个碎片.如果它超过2个进程，那么它可以限制源.

可用的流API动作包括

• ListStreams

• DescribeStream

• GetShardIterator

• GetRecords

您可以查看以下流读取和减去的示例;

import java.util.ArrayList;import java.util.HashMap;import java.util.List;import java.util.Map;import com.amazonaws.auth.profile.ProfileCredentialsProvider;import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient;import com.amazonaws.services.dynamodbv2.AmazonDynamoDBStreamsClient;import com.amazonaws.services.dynamodbv2.model.AttributeAction;import com.amazonaws.services.dynamodbv2.model.AttributeDefinition;import com.amazonaws.services.dynamodbv2.model.AttributeValue;import com.amazonaws.services.dynamodbv2.model.AttributeValueUpdate;import com.amazonaws.services.dynamodbv2.model.CreateTableRequest;import com.amazonaws.services.dynamodbv2.model.DescribeStreamRequest;import com.amazonaws.services.dynamodbv2.model.DescribeStreamResult;import com.amazonaws.services.dynamodbv2.model.DescribeTableResult;import com.amazonaws.services.dynamodbv2.model.GetRecordsRequest;import com.amazonaws.services.dynamodbv2.model.GetRecordsResult;import com.amazonaws.services.dynamodbv2.model.GetShardIteratorRequest;import com.amazonaws.services.dynamodbv2.model.GetShardIteratorResult;import com.amazonaws.services.dynamodbv2.model.KeySchemaElement;import com.amazonaws.services.dynamodbv2.model.KeyType;import com.amazonaws.services.dynamodbv2.model.ProvisionedThroughput;import com.amazonaws.services.dynamodbv2.model.Record;import com.amazonaws.services.dynamodbv2.model.Shard;import com.amazonaws.services.dynamodbv2.model.ShardIteratorType;import com.amazonaws.services.dynamodbv2.model.StreamSpecification;import com.amazonaws.services.dynamodbv2.model.StreamViewType;import com.amazonaws.services.dynamodbv2.util.Tables;public class StreamsExample {   private static AmazonDynamoDBClient dynamoDBClient =        new AmazonDynamoDBClient(new ProfileCredentialsProvider());     private static AmazonDynamoDBStreamsClient streamsClient =        new AmazonDynamoDBStreamsClient(new ProfileCredentialsProvider());     public static void main(String args[]) {        dynamoDBClient.setEndpoint("InsertDbEndpointHere");         streamsClient.setEndpoint("InsertStreamEndpointHere");                // table creation       String tableName = "MyTestingTable";        ArrayList attributeDefinitions =           new ArrayList();              attributeDefinitions.add(new AttributeDefinition()         .withAttributeName("ID")          .withAttributeType("N"));               ArrayList keySchema = new          ArrayList();             keySchema.add(new KeySchemaElement()          .withAttributeName("ID")          .withKeyType(KeyType.HASH));                       //Partition key      StreamSpecification streamSpecification = new StreamSpecification();       streamSpecification.setStreamEnabled(true);       streamSpecification.setStreamViewType(StreamViewType.NEW_AND_OLD_IMAGES);        CreateTableRequest createTableRequest = new CreateTableRequest()          .withTableName(tableName)          .withKeySchema(keySchema)          .withAttributeDefinitions(attributeDefinitions)          .withProvisionedThroughput(new ProvisionedThroughput()          .withReadCapacityUnits(1L)          .withWriteCapacityUnits(1L))         .withStreamSpecification(streamSpecification);              System.out.println("Executing CreateTable for " + tableName);       dynamoDBClient.createTable(createTableRequest);        System.out.println("Creating " + tableName);             try {          Tables.awaitTableToBecomeActive(dynamoDBClient, tableName);       } catch (InterruptedException e) {          e.printStackTrace();       }                // Get the table's stream settings       DescribeTableResult describeTableResult =         dynamoDBClient.describeTable(tableName);              String myStreamArn = describeTableResult.getTable().getLatestStreamArn();       StreamSpecification myStreamSpec =           describeTableResult.getTable().getStreamSpecification();              System.out.println("Current stream ARN for " + tableName + ": "+ myStreamArn);      System.out.println("Stream enabled: "+ myStreamSpec.getStreamEnabled());       System.out.println("Update view type: "+ myStreamSpec.getStreamViewType());              // Add an item       int numChanges = 0;       System.out.println("Making some changes to table data");       Map item = new HashMap();       item.put("ID", new AttributeValue().withN("222"));       item.put("Alert", new AttributeValue().withS("item!"));       dynamoDBClient.putItem(tableName, item);       numChanges++;              // Update the item               Map key = new HashMap();       key.put("ID", new AttributeValue().withN("222"));       Map attributeUpdates =        new HashMap();             attributeUpdates.put("Alert", new AttributeValueUpdate()          .withAction(AttributeAction.PUT)          .withValue(new AttributeValue().withS("modified item")));             dynamoDBClient.updateItem(tableName, key, attributeUpdates);       numChanges++;               // Delete the item               dynamoDBClient.deleteItem(tableName, key);        numChanges++;            // Get stream shards               DescribeStreamResult describeStreamResult =        streamsClient.describeStream(new DescribeStreamRequest()          .withStreamArn(myStreamArn));       String streamArn =           describeStreamResult.getStreamDescription().getStreamArn();       List shards =           describeStreamResult.getStreamDescription().getShards();              // Process shards       for (Shard shard : shards) {          String shardId = shard.getShardId();          System.out.println("Processing " + shardId + " in "+ streamArn);                    // Get shard iterator          GetShardIteratorRequest getShardIteratorRequest = new             GetShardIteratorRequest()             .withStreamArn(myStreamArn)             .withShardId(shardId)             .withShardIteratorType(ShardIteratorType.TRIM_HORIZON);                   GetShardIteratorResult getShardIteratorResult =              streamsClient.getShardIterator(getShardIteratorRequest);          String nextItr = getShardIteratorResult.getShardIterator();                    while (nextItr != null && numChanges > 0) {             // Read data records with iterator                             GetRecordsResult getRecordsResult =                 streamsClient.getRecords(new GetRecordsRequest().                withShardIterator(nextItr));                           List records = getRecordsResult.getRecords();             System.out.println("Pulling records...");                             for (Record record : records) {                System.out.println(record);                numChanges--;            }             nextItr = getRecordsResult.getNextShardIterator();          }       }    } }