Conceitos básicos do Kinesis Data Streams para o Amazon DynamoDB

Faça login no AWS Management Console e abra o console do Kinesis em https://console.aws.amazon.com/kinesis/.

Escolha Create data stream (Criar fluxo de dados) e siga as instruções para criar um fluxo chamado samplestream.

Abra o console do DynamoDB em https://console.aws.amazon.com/dynamodb/.

No painel de navegação, no lado esquerdo do console, selecione Tables (Tabelas).

Escolha a tabela Music.

Escolha a guia Exports and streams (Exportações e fluxos).

(Opcional) Em Detalhes do stream de dados do Amazon Kinesis, você pode alterar a precisão do carimbo de data e hora do registro de microssegundos (padrão) para milissegundos.

Escolha samplestream na lista suspensa.

Escolha o botão Ativar.

Criar um fluxo de dados do Kinesis denominado samplestream usando o comando create-stream.

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

Consulte Considerações sobre gerenciamento de fragmentos para o Kinesis Data Streams antes de definir o número de fragmentos para o fluxo de dados do Kinesis.

Verifique se o stream do Kinesis está ativo e pronto para uso usando o comando describe-stream.

aws kinesis describe-stream --stream-name samplestream

Habilite o streaming do Kinesis na tabela do DynamoDB usando o comando enable-kinesis-streaming-destination do DynamoDB. Substitua o valor de stream-arn pelo que foi retornado por describe-stream na etapa anterior. Opcionalmente, habilite o streaming com uma precisão mais granular (microssegundos) dos valores de carimbo de data e hora retornados em cada registro.

Habilite o streaming com precisão de carimbo de data e hora em microssegundos:

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

Ou habilite o streaming com a precisão padrão do carimbo de data e hora (milissegundos):

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

Verifique se o streaming do Kinesis está ativa na tabela usando o comando describe-kinesis-streaming-destination do DynamoDB.

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

Grave os dados na tabela do DynamoDB usando o comando put-item, conforme descrito no Guia do desenvolvedor do 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"} }'

Use o comando get-records da CLI do Kinesis para recuperar o conteúdo de fluxo do Kinesis. Em seguida, use o trecho de código a seguir para desserializar o conteúdo do fluxo.

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

Siga as instruções no Guia do desenvolvedor do Kinesis Data Streams para criar um fluxo de dados do Kinesis chamada samplestream usando Java.

Consulte Considerações sobre gerenciamento de fragmentos para o Kinesis Data Streams antes de definir o número de fragmentos para o fluxo de dados do Kinesis.

Use o trecho de código a seguir para habilitar a transmissão do Kinesis na tabela do DynamoDB. Opcionalmente, habilite o streaming com uma precisão mais granular (microssegundos) dos valores de carimbo de data e hora retornados em cada registro.

Habilite o streaming com precisão de carimbo de data e hora em microssegundos:

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

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

EnableKinesisStreamingDestinationResponse enableKdsResponse = ddbClient.enableKinesisStreamingDestination(enableKdsRequest);

Ou habilite o streaming com a precisão padrão do carimbo de data e hora (milissegundos):

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

EnableKinesisStreamingDestinationResponse enableKdsResponse = ddbClient.enableKinesisStreamingDestination(enableKdsRequest);

Siga as instruções do Kinesis Data Streams developer guide (Guia do desenvolvedor do Kinesis Data Streams) para ler o fluxo de dados criado.

Use o trecho de código a seguir para desserializar o conteúdo do fluxo

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