DynamoDB - Atividade de mesa

Os fluxos do DynamoDB permitem que você rastreie e responda às alterações dos itens da tabela. Empregue esta funcionalidade para criar um aplicativo que responde às mudanças atualizando as informações entre as fontes. Sincronize dados para milhares de usuários de um grande sistema multiusuário. Use-o para enviar notificações aos usuários sobre atualizações. Suas aplicações são diversas e substanciais. Os streams do DynamoDB servem como a principal ferramenta usada para atingir essa funcionalidade.

Os streams capturam sequências ordenadas por tempo contendo modificações de itens em uma tabela. Eles mantêm esses dados por no máximo 24 horas. Os aplicativos os usam para visualizar os itens originais e modificados, quase em tempo real.

Os fluxos ativados em uma tabela capturam todas as modificações. Em qualquer operação CRUD, o DynamoDB cria um registro de fluxo com os atributos de chave primária dos itens modificados. Você pode configurar fluxos para obter informações adicionais, como imagens anteriores e posteriores.

Os Streams trazem duas garantias -

Cada registro aparece uma vez no stream e
Cada modificação de item resulta nos registros de fluxo da mesma ordem das modificações.

Todos os fluxos são processados em tempo real para permitir que você os utilize para funcionalidades relacionadas em aplicativos.

Gerenciando Streams

Na criação da mesa, você pode habilitar um fluxo. As tabelas existentes permitem a desativação do fluxo ou alterações nas configurações. Streams oferecem o recurso de operação assíncrona, o que significa nenhum impacto no desempenho da tabela.

Utilize o console de gerenciamento da AWS para gerenciamento de fluxo simples. Primeiro, navegue até o console e escolhaTables. Na guia Visão geral, escolhaManage Stream. Dentro da janela, selecione as informações adicionadas a um fluxo nas modificações de dados da tabela. Depois de inserir todas as configurações, selecioneEnable.

Se você deseja desativar qualquer fluxo existente, selecione Manage Stream, e depois Disable.

Você também pode utilizar as APIs CreateTable e UpdateTable para habilitar ou alterar um fluxo. Use o parâmetro StreamSpecification para configurar o fluxo. StreamEnabled especifica o status, significando verdadeiro para habilitado e falso para desabilitado.

StreamViewType especifica as informações adicionadas ao fluxo: KEYS_ONLY, NEW_IMAGE, OLD_IMAGE e NEW_AND_OLD_IMAGES.

Leitura de fluxo

Leia e processe fluxos conectando-se a um terminal e fazendo solicitações de API. Cada fluxo consiste em registros de fluxo e cada registro existe como uma única modificação que possui o fluxo. Os registros de fluxo incluem um número de sequência que revela a ordem de publicação. Os registros pertencem a grupos também conhecidos como shards. Os shards funcionam como contêineres para vários registros e também contêm informações necessárias para acessar e percorrer os registros. Após 24 horas, os registros são excluídos automaticamente.

Esses fragmentos são gerados e excluídos conforme necessário e não duram muito. Eles também se dividem em vários novos shards automaticamente, geralmente em resposta a picos de atividade de gravação. Desativando na transmissão, feche os fragmentos abertos. O relacionamento hierárquico entre os shards significa que os aplicativos devem priorizar os shards pais para a ordem de processamento correta. Você pode usar o adaptador Kinesis para fazer isso automaticamente.

Note - As operações que resultam em nenhuma mudança não gravam registros de fluxo.

Para acessar e processar registros, é necessário realizar as seguintes tarefas -

Determine o ARN do fluxo de destino.
Determine o (s) fragmento (s) do fluxo que contém os registros de destino.
Acesse o (s) fragmento (s) para recuperar os registros desejados.

Note- Deve haver no máximo 2 processos lendo um fragmento de uma vez. Se exceder 2 processos, pode estrangular a fonte.

As ações da API de fluxo disponíveis incluem

ListStreams
DescribeStream
GetShardIterator
GetRecords

Você pode revisar o seguinte exemplo de leitura de fluxo -

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.amazonaws.auth.profile.ProfileCredentialsProvider;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBStreamsClient;

import com.amazonaws.services.dynamodbv2.model.AttributeAction;
import com.amazonaws.services.dynamodbv2.model.AttributeDefinition;
import com.amazonaws.services.dynamodbv2.model.AttributeValue;
import com.amazonaws.services.dynamodbv2.model.AttributeValueUpdate;

import com.amazonaws.services.dynamodbv2.model.CreateTableRequest;
import com.amazonaws.services.dynamodbv2.model.DescribeStreamRequest;
import com.amazonaws.services.dynamodbv2.model.DescribeStreamResult;
import com.amazonaws.services.dynamodbv2.model.DescribeTableResult;

import com.amazonaws.services.dynamodbv2.model.GetRecordsRequest;
import com.amazonaws.services.dynamodbv2.model.GetRecordsResult;
import com.amazonaws.services.dynamodbv2.model.GetShardIteratorRequest;
import com.amazonaws.services.dynamodbv2.model.GetShardIteratorResult;

import com.amazonaws.services.dynamodbv2.model.KeySchemaElement;
import com.amazonaws.services.dynamodbv2.model.KeyType;
import com.amazonaws.services.dynamodbv2.model.ProvisionedThroughput;
import com.amazonaws.services.dynamodbv2.model.Record;

import com.amazonaws.services.dynamodbv2.model.Shard;
import com.amazonaws.services.dynamodbv2.model.ShardIteratorType;
import com.amazonaws.services.dynamodbv2.model.StreamSpecification;
import com.amazonaws.services.dynamodbv2.model.StreamViewType;
import com.amazonaws.services.dynamodbv2.util.Tables;

public class StreamsExample {
   private static AmazonDynamoDBClient dynamoDBClient =  
      new AmazonDynamoDBClient(new ProfileCredentialsProvider());  
   private static AmazonDynamoDBStreamsClient streamsClient =  
      new AmazonDynamoDBStreamsClient(new ProfileCredentialsProvider());  

   public static void main(String args[]) {  
      dynamoDBClient.setEndpoint("InsertDbEndpointHere");   
      streamsClient.setEndpoint("InsertStreamEndpointHere");    
      
      // table creation 
      String tableName = "MyTestingTable";  
      ArrayList<AttributeDefinition> attributeDefinitions =  
         new ArrayList<AttributeDefinition>();  
      
      attributeDefinitions.add(new AttributeDefinition()
         .withAttributeName("ID") 
         .withAttributeType("N"));
         
      ArrayList<KeySchemaElement> keySchema = new 
         ArrayList<KeySchemaElement>(); 
      
      keySchema.add(new KeySchemaElement() 
         .withAttributeName("ID") 
         .withKeyType(KeyType.HASH));                       //Partition key

      StreamSpecification streamSpecification = new StreamSpecification(); 
      streamSpecification.setStreamEnabled(true); 
      streamSpecification.setStreamViewType(StreamViewType.NEW_AND_OLD_IMAGES);  
      CreateTableRequest createTableRequest = new CreateTableRequest() 
         .withTableName(tableName) 
         .withKeySchema(keySchema) 
         .withAttributeDefinitions(attributeDefinitions) 
         .withProvisionedThroughput(new ProvisionedThroughput() 
         .withReadCapacityUnits(1L) 
         .withWriteCapacityUnits(1L))
         .withStreamSpecification(streamSpecification);  
      
      System.out.println("Executing CreateTable for " + tableName); 
      dynamoDBClient.createTable(createTableRequest);  
      System.out.println("Creating " + tableName); 
      
      try { 
         Tables.awaitTableToBecomeActive(dynamoDBClient, tableName); 
      } catch (InterruptedException e) { 
         e.printStackTrace(); 
      } 
         
      // Get the table's stream settings 
      DescribeTableResult describeTableResult =
         dynamoDBClient.describeTable(tableName);  
      
      String myStreamArn = describeTableResult.getTable().getLatestStreamArn(); 
      StreamSpecification myStreamSpec =  
         describeTableResult.getTable().getStreamSpecification();  
      
      System.out.println("Current stream ARN for " + tableName + ": "+ myStreamArn);
      System.out.println("Stream enabled: "+ myStreamSpec.getStreamEnabled()); 
      System.out.println("Update view type: "+ myStreamSpec.getStreamViewType());  
      
      // Add an item 
      int numChanges = 0; 
      System.out.println("Making some changes to table data"); 
      Map<String, AttributeValue> item = new HashMap<String, AttributeValue>(); 
      item.put("ID", new AttributeValue().withN("222")); 
      item.put("Alert", new AttributeValue().withS("item!")); 
      dynamoDBClient.putItem(tableName, item); 
      numChanges++;  
      
      // Update the item         
      Map<String, AttributeValue> key = new HashMap<String, AttributeValue>(); 
      key.put("ID", new AttributeValue().withN("222")); 
      Map<String, AttributeValueUpdate> attributeUpdates =  
      new HashMap<String, AttributeValueUpdate>(); 
      
      attributeUpdates.put("Alert", new AttributeValueUpdate() 
         .withAction(AttributeAction.PUT) 
         .withValue(new AttributeValue().withS("modified item"))); 
      
      dynamoDBClient.updateItem(tableName, key, attributeUpdates); 
      numChanges++;   
      
      // Delete the item         
      dynamoDBClient.deleteItem(tableName, key);  
      numChanges++;
      
      // Get stream shards         
      DescribeStreamResult describeStreamResult =  
      streamsClient.describeStream(new DescribeStreamRequest() 
         .withStreamArn(myStreamArn)); 
      String streamArn =  
         describeStreamResult.getStreamDescription().getStreamArn(); 
      List<Shard> shards =  
         describeStreamResult.getStreamDescription().getShards();  
      
      // Process shards 
      for (Shard shard : shards) { 
         String shardId = shard.getShardId(); 
         System.out.println("Processing " + shardId + " in "+ streamArn);  
         
         // Get shard iterator 
         GetShardIteratorRequest getShardIteratorRequest = new 
            GetShardIteratorRequest() 
            .withStreamArn(myStreamArn) 
            .withShardId(shardId) 
            .withShardIteratorType(ShardIteratorType.TRIM_HORIZON); 
         
         GetShardIteratorResult getShardIteratorResult =  
            streamsClient.getShardIterator(getShardIteratorRequest); 
         String nextItr = getShardIteratorResult.getShardIterator();  
         
         while (nextItr != null && numChanges > 0) { 
            // Read data records with iterator                 
            GetRecordsResult getRecordsResult =  
               streamsClient.getRecords(new GetRecordsRequest(). 
               withShardIterator(nextItr));
               
            List<Record> records = getRecordsResult.getRecords(); 
            System.out.println("Pulling records...");  
               
            for (Record record : records) { 
               System.out.println(record); 
               numChanges--;
            } 
            nextItr = getRecordsResult.getNextShardIterator(); 
         } 
      } 
   } 
}