Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

# Memulai Kinesis Data Streams untuk Amazon DynamoDB
<a name="kds_gettingstarted"></a>

Bagian ini menjelaskan cara menggunakan Kinesis Data Streams untuk tabel Amazon DynamoDB dengan konsol Amazon AWS Command Line Interface AWS CLI DynamoDB, (), dan API.

## Membuat aliran data Amazon Kinesis yang aktif
<a name="kds_gettingstarted.making-changes"></a>

Semua contoh ini menggunakan tabel `Music` DynamoDB yang dibuat sebagai bagian dari tutorial [Memulai dengan DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/GettingStartedDynamoDB.html).

Untuk mempelajari lebih lanjut tentang cara membangun konsumen dan menghubungkan aliran data Kinesis Anda ke layanan AWS lain, lihat [Membaca data dari Kinesis Data Streams](https://docs.aws.amazon.com/streams/latest/dev/building-consumers.html) dalam *panduan pengembang Amazon Kinesis Data Streams*.

**catatan**  
 Saat Anda pertama kali menggunakan serpihan KDS, sebaiknya atur serpihan Anda untuk ditingkatkan dan diturunkan skalanya sesuai dengan pola penggunaan. Setelah mengumpulkan lebih banyak data tentang pola penggunaan, Anda dapat menyesuaikan serpihan di aliran agar sesuai. 

------
#### [ Console ]

1. Masuk ke Konsol Manajemen AWS dan buka konsol Kinesis di. [https://console.aws.amazon.com/kinesis/](https://console.aws.amazon.com/kinesis/)

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

1. Buka konsol DynamoDB di. [https://console.aws.amazon.com/dynamodb/](https://console.aws.amazon.com/dynamodb/)

1. Di panel navigasi di sisi kiri konsol, pilih **Tabel**.

1. Pilih tabel **Musik**.

1. Pilih tab **Ekspor dan aliran**.

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

1. Pilih **samplestream** dari daftar menurun.

1. Pilih tombol **Hidupkan**.

------
#### [ AWS CLI ]

1. Buat Kinesis Data Streams bernama `samplestream` dengan menggunakan [perintah create-stream](https://docs.aws.amazon.com/cli/latest/reference/kinesis/create-stream.html).

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

   Lihat [Pertimbangan manajemen pecahan untuk Kinesis Data Streams](kds_using-shards-and-metrics.md#kds_using-shards-and-metrics.shardmanagment) sebelum menetapkan jumlah serpihan untuk Kinesis data stream.

1. Periksa apakah stream Kinesis aktif dan siap digunakan dengan menggunakan [perintah describe-stream](https://docs.aws.amazon.com/cli/latest/reference/kinesis/describe-stream.html).

   ```
   aws kinesis describe-stream --stream-name samplestream
   ```

1. 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
   ```

1. 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
   ```

1. Menulis data ke Daftar Tabel DynamoDB dengan menggunakan perintah `put-item`, seperti yang dijelaskan dalam [Panduan Developer DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/getting-started-step-2.html).

   ```
   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"} }'
   ```

1. Gunakan perintah CLI Kinesis [get-records](https://docs.aws.amazon.com/cli/latest/reference/kinesis/get-records.html) untuk mengambil konten stream 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());
   }
   ```

------
#### [ Java ]

1. Ikuti petunjuk dalam panduan developer Kinesis Data Streams untuk [menciptakan](https://docs.aws.amazon.com/streams/latest/dev/kinesis-using-sdk-java-create-stream.html) Kinesis data stream bernama `samplestream` menggunakan Java.

   Lihat [Pertimbangan manajemen pecahan untuk Kinesis Data Streams](kds_using-shards-and-metrics.md#kds_using-shards-and-metrics.shardmanagment) sebelum menetapkan jumlah serpihan untuk Kinesis data stream. 

1. 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);
   ```

1. Ikuti instruksi dalam panduan pengembang *Kinesis Data Streams* untuk [membaca](https://docs.aws.amazon.com/streams/latest/dev/building-consumers.html) dari aliran data yang dibuat.

1. 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());
   }
   ```

------

## Membuat perubahan pada aliran data Amazon Kinesis yang aktif
<a name="kds_gettingstarted.making-changes"></a>

Bagian ini menjelaskan cara membuat perubahan pada Kinesis Data Streams aktif untuk penyiapan DynamoDB dengan menggunakan konsol, dan API. AWS CLI 

**Konsol Manajemen AWS**

1. Buka konsol DynamoDB di [https://console.aws.amazon.com/dynamodb/](https://console.aws.amazon.com/dynamodb/)

1. Pergi ke meja Anda.

1. Pilih **Ekspor dan Aliran**.

**AWS CLI**

1. Hubungi `describe-kinesis-streaming-destination` untuk mengonfirmasi bahwa streaming tersebut`ACTIVE`. 

1. Panggilan`UpdateKinesisStreamingDestination`, seperti dalam contoh ini:

   ```
   aws dynamodb update-kinesis-streaming-destination --table-name enable_test_table --stream-arn arn:aws:kinesis:us-east-1:12345678901:stream/enable_test_stream --update-kinesis-streaming-configuration ApproximateCreationDateTimePrecision=MICROSECOND
   ```

1. Hubungi `describe-kinesis-streaming-destination` untuk mengonfirmasi bahwa streaming tersebut`UPDATING`.

1. Panggil `describe-kinesis-streaming-destination` secara berkala hingga status streaming `ACTIVE` kembali. Biasanya diperlukan waktu hingga 5 menit agar pembaruan presisi stempel waktu diterapkan. Setelah status ini diperbarui, itu menunjukkan bahwa pembaruan selesai dan nilai presisi baru akan diterapkan pada catatan masa depan.

1. Tulis ke meja menggunakan`putItem`.

1. Gunakan `get-records` perintah Kinesis untuk mendapatkan konten aliran.

1. Konfirmasikan bahwa penulisan memiliki presisi yang diinginkan. `ApproximateCreationDateTime`

**API Java**

1. Berikan cuplikan kode yang membuat `UpdateKinesisStreamingDestination` permintaan dan respons. `UpdateKinesisStreamingDestination` 

1. Berikan cuplikan kode yang membuat `DescribeKinesisStreamingDestination` permintaan dan file. `DescribeKinesisStreamingDestination response`

1. Panggil `describe-kinesis-streaming-destination` secara berkala hingga status streaming `ACTIVE` kembali, menunjukkan bahwa pembaruan selesai dan nilai presisi baru akan diterapkan pada catatan masa depan.

1. Lakukan menulis ke meja.

1.  Baca dari aliran dan deserialisasi konten streaming.

1. Konfirmasikan bahwa penulisan memiliki presisi yang diinginkan. `ApproximateCreationDateTime`