Amazon DynamoDB 專用 Kinesis Data Streams 入門

登入 AWS Management Console ，並在 開啟 Kinesis 主控台https://console.aws.amazon.com/kinesis/。

選擇 Create data stream (建立資料串流)，並依照說明來建立名為 samplestream 的串流。

在 開啟 DynamoDB 主控台https://console.aws.amazon.com/dynamodb/。

在主控台左側的導覽窗格中，選擇 Tables (資料表)。

選擇 Music (音樂) 資料表。

選擇 Exports and streams (匯出與串流) 索引標籤。

（選用） 在 Amazon Kinesis 資料串流詳細資訊下，您可以將記錄時間戳記精確度從微秒 （預設） 變更為毫秒。

從下拉式清單選擇 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

使用 put-item 命令將資料寫入 DynamoDB 資料表，如《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 的 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());
}