适用于 Amazon DynamoDB 的 Kinesis Data Streams 入门

登录到 AWS Management Console，然后打开 Kinesis 控制台：https://console.aws.amazon.com/kinesis/。

选择创建数据流并按照说明创建一个名为 samplestream 的流。

打开 DynamoDB 控制台：https://console.aws.amazon.com/dynamodb/。

在控制台左侧的导航窗格中，选择表。

选择 Music 表。

选择导出和流式传输选项卡。

（可选）在 Amazon Kinesis Data Streams 详细信息下，您可以将记录时间戳精度从微秒（默认）更改为毫秒。

从下拉列表选择 samplestream。

选择开启按钮。

使用 create-stream 命令创建名为 samplestream 的 Kinesis 流。

aws kinesis create-stream --stream-name samplestream --shard-count 3 

请参阅Kinesis Data Streams 的分片管理注意事项，然后设置 Kinesis 数据流的分片数。

使用 describe-stream 命令，检查 Kinesis 流是否处于活动状态并准备好使用。

aws kinesis describe-stream --stream-name samplestream

使用 DynamoDB enable-kinesis-streaming-destination 命令，在 DynamoDB 表上启用 Kinesis 数据流。将 stream-arn 值替换为上一步返回的 describe-stream。（可选）启用在每条记录上返回更精细（微秒）精度时间戳值的流式传输。

启用微秒时间戳精度的流式传输：

aws dynamodb enable-kinesis-streaming-destination \
  --table-name Music \
  --stream-arn arn:aws:kinesis:us-west-2:12345678901:stream/samplestream
  --enable-kinesis-streaming-configuration ApproximateCreationDateTimePrecision=MICROSECOND

或者启用默认时间戳精度（毫秒）的流式传输：

aws dynamodb enable-kinesis-streaming-destination \
  --table-name Music \
  --stream-arn arn:aws:kinesis:us-west-2:12345678901:stream/samplestream

使用 DynamoDB describe-kinesis-streaming-destination 命令检查是否在表上激活 Kinesis 流。

aws dynamodb describe-kinesis-streaming-destination --table-name Music

使用 DynamoDB 开发人员指南介绍的 put-item 命令，将数据写入 DynamoDB 表。

aws dynamodb put-item \
    --table-name Music  \
    --item \
        '{"Artist": {"S": "No One You Know"}, "SongTitle": {"S": "Call Me Today"}, "AlbumTitle": {"S": "Somewhat Famous"}, "Awards": {"N": "1"}}'

aws dynamodb put-item \
    --table-name Music \
    --item \
        '{"Artist": {"S": "Acme Band"}, "SongTitle": {"S": "Happy Day"}, "AlbumTitle": {"S": "Songs About Life"}, "Awards": {"N": "10"} }'

使用 Kinesis get-records CLI 命令检索 Kinesis 流内容。然后使用以下代码片段来反序列化流内容。

/**
 * Takes as input a Record fetched from Kinesis and does arbitrary processing as an example.
 */
public void processRecord(Record kinesisRecord) throws IOException {
    ByteBuffer kdsRecordByteBuffer = kinesisRecord.getData();
    JsonNode rootNode = OBJECT_MAPPER.readTree(kdsRecordByteBuffer.array());
    JsonNode dynamoDBRecord = rootNode.get("dynamodb");
    JsonNode oldItemImage = dynamoDBRecord.get("OldImage");
    JsonNode newItemImage = dynamoDBRecord.get("NewImage");
    Instant recordTimestamp = fetchTimestamp(dynamoDBRecord);

    /**
     * Say for example our record contains a String attribute named "stringName" and we want to fetch the value
     * of this attribute from the new item image. The following code fetches this value.
     */
    JsonNode attributeNode = newItemImage.get("stringName");
    JsonNode attributeValueNode = attributeNode.get("S"); // Using DynamoDB "S" type attribute
    String attributeValue = attributeValueNode.textValue();
    System.out.println(attributeValue);
}

private Instant fetchTimestamp(JsonNode dynamoDBRecord) {
    JsonNode timestampJson = dynamoDBRecord.get("ApproximateCreationDateTime");
    JsonNode timestampPrecisionJson = dynamoDBRecord.get("ApproximateCreationDateTimePrecision");
    if (timestampPrecisionJson != null && timestampPrecisionJson.equals("MICROSECOND")) {
        return Instant.EPOCH.plus(timestampJson.longValue(), ChronoUnit.MICROS);
    }
    return Instant.ofEpochMilli(timestampJson.longValue());
}

按照 Kinesis Data Streams 开发人员指南中的说明，使用 Java 创建一个名为 samplestream 的https://docs.aws.amazon.com/streams/latest/dev/kinesis-using-sdk-java-create-stream.html Kinesis 数据流。

请参阅Kinesis Data Streams 的分片管理注意事项，然后设置 Kinesis 数据流的分片数。

使用以下代码段在DynamoDB 表上启用 Kinesis 数据流。（可选）启用在每条记录上返回更精细（微秒）精度时间戳值的流式传输。

启用微秒时间戳精度的流式传输：

EnableKinesisStreamingConfiguration enableKdsConfig = EnableKinesisStreamingConfiguration.builder()
  .approximateCreationDateTimePrecision(ApproximateCreationDateTimePrecision.MICROSECOND)
  .build();

EnableKinesisStreamingDestinationRequest enableKdsRequest = EnableKinesisStreamingDestinationRequest.builder()
  .tableName(tableName)
  .streamArn(kdsArn)
  .enableKinesisStreamingConfiguration(enableKdsConfig)
  .build();

EnableKinesisStreamingDestinationResponse enableKdsResponse = ddbClient.enableKinesisStreamingDestination(enableKdsRequest);

或者启用默认时间戳精度（毫秒）的流式传输：

EnableKinesisStreamingDestinationRequest enableKdsRequest = EnableKinesisStreamingDestinationRequest.builder()
  .tableName(tableName)
  .streamArn(kdsArn)
  .build();

EnableKinesisStreamingDestinationResponse enableKdsResponse = ddbClient.enableKinesisStreamingDestination(enableKdsRequest);

按照 Kinesis Data Streams 开发人员指南中的说明进行操作，从创建的数据流读取。

然后使用以下代码段来反序列化流内容。

/**
 * Takes as input a Record fetched from Kinesis and does arbitrary processing as an example.
 */
public void processRecord(Record kinesisRecord) throws IOException {
    ByteBuffer kdsRecordByteBuffer = kinesisRecord.getData();
    JsonNode rootNode = OBJECT_MAPPER.readTree(kdsRecordByteBuffer.array());
    JsonNode dynamoDBRecord = rootNode.get("dynamodb");
    JsonNode oldItemImage = dynamoDBRecord.get("OldImage");
    JsonNode newItemImage = dynamoDBRecord.get("NewImage");
    Instant recordTimestamp = fetchTimestamp(dynamoDBRecord);

    /**
     * Say for example our record contains a String attribute named "stringName" and we wanted to fetch the value
     * of this attribute from the new item image, the below code would fetch this.
     */
    JsonNode attributeNode = newItemImage.get("stringName");
    JsonNode attributeValueNode = attributeNode.get("S"); // Using DynamoDB "S" type attribute
    String attributeValue = attributeValueNode.textValue();
    System.out.println(attributeValue);
}

private Instant fetchTimestamp(JsonNode dynamoDBRecord) {
    JsonNode timestampJson = dynamoDBRecord.get("ApproximateCreationDateTime");
    JsonNode timestampPrecisionJson = dynamoDBRecord.get("ApproximateCreationDateTimePrecision");
    if (timestampPrecisionJson != null && timestampPrecisionJson.equals("MICROSECOND")) {
        return Instant.EPOCH.plus(timestampJson.longValue(), ChronoUnit.MICROS);
    }
    return Instant.ofEpochMilli(timestampJson.longValue());
}