Memulai Kinesis Data Streams untuk Amazon DynamoDB

Masuk ke AWS Management Console dan buka konsol Kinesis di. https://console.aws.amazon.com/kinesis/

Pilih Buat aliran data dan ikuti petunjuk untuk membuat stream yang disebut samplestream.

Buka konsol DynamoDB di. https://console.aws.amazon.com/dynamodb/

Di panel navigasi di sisi kiri konsol, pilih Tabel.

Pilih tabel Musik.

Pilih tab Ekspor dan aliran.

(Opsional) Di bawah detail aliran data Amazon Kinesis, Anda dapat mengubah presisi stempel waktu rekaman dari mikrodetik (default) menjadi milidetik.

Pilih samplestream dari daftar menurun.

Pilih tombol Hidupkan.

Buat Kinesis Data Streams bernama samplestream dengan menggunakan perintah create-stream.

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

Lihat Pertimbangan manajemen pecahan untuk Kinesis Data Streams sebelum menetapkan jumlah serpihan untuk Kinesis data stream.

Periksa apakah stream Kinesis aktif dan siap digunakan dengan menggunakan perintah describe-stream.

aws kinesis describe-stream --stream-name samplestream

Aktifkan Kinesis streaming pada tabel DynamoDB dengan menggunakan perintah enable-kinesis-streaming-destination DynamoDB. Ganti stream-arn nilai dengan yang ditampilkan describe-stream pada langkah sebelumnya. Secara opsional, aktifkan streaming dengan presisi nilai stempel waktu yang lebih granular (mikrodetik) yang dikembalikan pada setiap catatan.

Aktifkan streaming dengan presisi stempel waktu mikrodetik:

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

Atau aktifkan streaming dengan presisi stempel waktu default (milidetik):

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

Periksa apakah streaming Kinesis aktif pada tabel DynamoDB dengan menggunakan perintah describe-kinesis-streaming-destination DynamoDB.

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

Menulis data ke Daftar Tabel DynamoDB dengan menggunakan perintah put-item, seperti yang dijelaskan dalam Panduan Developer 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"} }'

Gunakan CLI perintah Kinesis get-records untuk mengambil konten aliran Kinesis. Kemudian gunakan potongan kode berikut untuk melakukan deserialisasi konten stream.

/**
 * 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());
}

Ikuti petunjuk dalam panduan developer Kinesis Data Streams untuk menciptakan Kinesis data stream bernama samplestream menggunakan Java.

Lihat Pertimbangan manajemen pecahan untuk Kinesis Data Streams sebelum menetapkan jumlah serpihan untuk Kinesis data stream.

Gunakan cuplikan kode berikut untuk mengaktifkan Kinesis streaming pada tabel DynamoDB. Secara opsional, aktifkan streaming dengan presisi nilai stempel waktu yang lebih granular (mikrodetik) yang dikembalikan pada setiap catatan.

Aktifkan streaming dengan presisi stempel waktu mikrodetik:

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

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

EnableKinesisStreamingDestinationResponse enableKdsResponse = ddbClient.enableKinesisStreamingDestination(enableKdsRequest);

Atau aktifkan streaming dengan presisi stempel waktu default (milidetik):

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

EnableKinesisStreamingDestinationResponse enableKdsResponse = ddbClient.enableKinesisStreamingDestination(enableKdsRequest);

Ikuti instruksi dalam panduan pengembang Kinesis Data Streams untuk membaca dari aliran data yang dibuat.

Gunakan potongan kode berikut untuk melakukan deserialisasi konten stream

/**
 * 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());
}