Nozioni di base su Kinesis Data Streams per Amazon DynamoDB

Accedi AWS Management Console e apri la console Kinesis all'indirizzo. https://console.aws.amazon.com/kinesis/

Scegli Crea flusso di dati e segui le istruzioni per creare un flusso denominato samplestream.

Apri la console DynamoDB all'indirizzo. https://console.aws.amazon.com/dynamodb/

Nel riquadro di navigazione sul lato sinistro della console scegli Tables (Tabelle).

Seleziona la tabella Music.

Scegli la scheda Exports and streams (Esportazioni e flussi).

(Facoltativo) Nella sezione Dettagli del flusso di dati di Amazon Kinesis, puoi modificare la precisione del timestamp del record da microsecondi (impostazione predefinita) a millisecondi.

Scegli samplestream dall'elenco a discesa.

Scegli il pulsante Attiva.

Crea un flusso di dati Kinesis denominato samplestream utilizzando il comando create-stream.

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

Prima di configurare il numero di partizioni per il flusso di dati Kinesis, consulta Considerazioni sulla gestione delle partizioni per Kinesis Data Streams.

Verificare che il flusso Kinesis sia attivo e pronto per l'uso utilizzando il comando describe-stream.

aws kinesis describe-stream --stream-name samplestream

Abilitare lo streaming di Kinesis sulla tabella DynamoDB utilizzando il comando DynamoDB enable-kinesis-streaming-destination. Sostituisci il valore stream-arn con quello restituito da describe-stream nella fase precedente. Facoltativamente, abilita lo streaming con una precisione più granulare (microsecondi) dei valori di timestamp restituiti su ogni record.

Abilita lo streaming con una precisione di timestamp in microsecondi:

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

Oppure abilita lo streaming con la precisione del timestamp predefinita (millisecondi):

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

Abilitare lo streaming di Kinesis sulla tabella DynamoDB utilizzando il comando describe-kinesis-streaming-destination DynamoDB.

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

Scrivere i dati nella tabella DynamoDB utilizzando il comando put-item, come descritto nella Guida per gli sviluppatori di 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"} }'

Utilizzate il CLI comando Kinesis get-records per recuperare il contenuto del flusso Kinesis. Quindi utilizzare il seguente frammento di codice per deserializzare il contenuto del flusso.

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

Seguire le istruzioni contenute nella guida per gli sviluppatori di Kinesis Data Streams per creare un flusso di dati Kinesis denominato samplestream tramite Java.

Prima di configurare il numero di partizioni per il flusso di dati Kinesis, consulta Considerazioni sulla gestione delle partizioni per Kinesis Data Streams.

Utilizzo il seguente frammento di codice per abilitare lo streamingKinesis sulla tabella DynamoDB. Facoltativamente, abilita lo streaming con una precisione più granulare (microsecondi) dei valori di timestamp restituiti su ogni record.

Abilita lo streaming con una precisione di timestamp in microsecondi:

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

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

EnableKinesisStreamingDestinationResponse enableKdsResponse = ddbClient.enableKinesisStreamingDestination(enableKdsRequest);

Oppure abilita lo streaming con la precisione del timestamp predefinita (millisecondi):

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

EnableKinesisStreamingDestinationResponse enableKdsResponse = ddbClient.enableKinesisStreamingDestination(enableKdsRequest);

Segui le istruzioni contenute nella guida per gli sviluppatori di Kinesis Data Streams per leggere dal flusso di dati creato.

Utilizza il seguente frammento di codice per deserializzare il contenuto del flusso.

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