Amazon DynamoDB 專用 Kinesis Data Streams 入門

登入 AWS 管理主控台 並開啟位於 https：//https://console.aws.amazon.com/kinesis/ 的 Kinesis 主控台。

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

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

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

選擇 Music (音樂) 資料表。

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

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

從下拉式清單選擇 samplestream (範例串流)。

選擇 Turn On (開啟) 按鈕。

使用 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());
}