Usar um pipeline do OpenSearch Ingestion com o Amazon DynamoDB

Você pode usar o plug-in do DynamoDB para transmitir continuamente eventos de tabela, como criações, atualizações e exclusões, para domínios do Amazon OpenSearch Service e coleções do Amazon OpenSearch sem Servidor. O pipeline usa captura de dados alterados (CDC) para transmissão contínua em grande escala e baixa latência.

Você pode processar dados do DynamoDB com ou sem um snapshot inicial completo.

Para obter mais informações, consulte Integração ETL zero do DynamoDB com o Amazon OpenSearch Service no Guia do desenvolvedor do Amazon DynamoDB.

Pré-requisitos

Para configurar o pipeline, você precisa ter uma tabela do DynamoDB com o DynamoDB Streams habilitado. Seu fluxo deve usar o tipo de visualização de fluxo NEW_IMAGE. Porém, os pipelines do OpenSearch Ingestion também podem transmitir eventos com NEW_AND_OLD_IMAGES se esse tipo de visualização de fluxo for apropriado para o seu caso de uso.

Se estiver usando snapshots, você também deverá habilitar a recuperação para um ponto no tempo na sua tabela. Para obter mais informações, consulte Criar uma tabela, Habilitar a recuperação para um ponto no tempo e Habilitar um fluxo, no Guia do Desenvolvedor do Amazon DynamoDB.

Etapa 1: configurar a função do pipeline

Depois de configurar a tabela do DynamoDB, defina o perfil de pipeline que você deseja usar na configuração do pipeline e adicione as seguintes permissões do DynamoDB nesse perfil:

Também é possível usar uma chave gerenciada pelo cliente do AWS KMS para criptografar os arquivos de dados de exportação. Para descriptografar os objetos exportados, especifique s3_sse_kms_key_id para o ID da chave na configuração de exportação do pipeline, com o seguinte formato: arn:aws:kms:region:account-id:key/my-key-id. A política a seguir inclui as permissões necessárias para usar uma chave gerenciada pelo cliente:

{
    "Sid": "allowUseOfCustomManagedKey",
    "Effect": "Allow",
    "Action": [
        "kms:GenerateDataKey",
        "kms:Decrypt"
    ],
    "Resource": arn:aws:kms:region:account-id:key/my-key-id
}

Etapa 2: Criar o pipeline

Em seguida, é possível configurar um pipeline da Ingestão do OpenSearch como o seguinte, que especifica o DynamoDB como origem. Essa amostra de pipeline ingere dados de table-a com o snapshot de PITR, seguido por eventos do DynamoDB Streams. Uma posição inicial de LATEST indica que o pipeline deve ler os dados mais recentes do DynamoDB Streams.

version: "2"
cdc-pipeline:
  source:
    dynamodb:
      tables:
      - table_arn: "arn:aws:dynamodb:region:account-id:table/table-a"  
        export:
          s3_bucket: "my-bucket"
          s3_prefix: "export/"
        stream:
          start_position: "LATEST"
      aws:
        region: "us-east-1"
  sink:
  - opensearch:
      hosts: ["https://search-mydomain.region.es.amazonaws.com"]
      index: "${getMetadata(\"table-name\")}"
      index_type: custom
      normalize_index: true
      document_id: "${getMetadata(\"primary_key\")}"
      action: "${getMetadata(\"opensearch_action\")}"
      document_version: "${getMetadata(\"document_version\")}"
      document_version_type: "external"

Você pode usar um esquema do DynamoDB pré-configurado para criar esse pipeline. Para obter mais informações, consulte Trabalhar com esquemas.

Consistência de dados

O OpenSearch Ingestion oferece suporte ao reconhecimento de ponta a ponta para garantir a durabilidade dos dados. Quando um pipeline lê snapshots ou fluxos, ele cria partições dinamicamente para processamento paralelo. O pipeline marca uma partição como concluída quando recebe uma confirmação depois de ingerir todos os registros no domínio ou na coleção do OpenSearch.

Se você quiser fazer a ingestão em uma coleção de pesquisa do OpenSearch sem Servidor, gere um ID de documento no pipeline. Se quiser fazer a ingestão em uma coleção de séries temporais do OpenSearch sem Servidor, observe que o pipeline não gera um ID de documento.

Um pipeline do OpenSearch Ingestion também mapeia ações de eventos de entrada para ações de indexação em massa correspondentes para ajudar a ingerir documentos. Isso mantém os dados consistentes, de modo que cada alteração de dados no DynamoDB seja reconciliada com as alterações correspondentes do documento no OpenSearch.

Mapear tipo de dados

O OpenSearch Service mapeia tipos de dados dinamicamente em cada documento recebido para o tipo de dados correspondente no DynamoDB. A tabela a seguir mostra como o OpenSearch Service mapeia automaticamente vários tipos de dados.

{
 "template": {
  "mappings": {
   "properties": {
    "MixedNumberAttribute": {
     "type": "float"
    }
   }
  }
 }
}

{
 "template": {
  "mappings": {
   "properties": {
    "PartType": {
     "type": "keyword"
    }
   }
  }
 }
}

{
 "template": {
  "mappings": {
   "properties": {
    "ImageData": {
     "type": "binary"
    }
   }
  }
 }
}

{
 "template": {
  "mappings": {
   "properties": {
    "AdditionalDescriptions": {
     "properties": {
      "PartType": {
       "type": "keyword"
      }
     }
    }
   }
  }
 }
}

Recomendamos que você configure a fila de mensagens não entregues (DLQ) no pipeline do OpenSearch Ingestion. Se você tiver configurado a fila, o OpenSearch Service enviará todos os documentos com falha que não puderem ser ingeridos devido a falhas de mapeamento dinâmico para a fila.

Se os mapeamentos automáticos falharem, será possível usar template_type e template_content na configuração do pipeline para definir regras de mapeamento explícitas. Como alternativa, é possível criar modelos de mapeamento diretamente no seu domínio de pesquisa ou na sua coleção antes de iniciar o pipeline.

Limitações

Considere as seguintes limitações ao configurar um pipeline do OpenSearch Ingestion para o DynamoDB:

Alarmes do CloudWatch recomendados para DynamoDB

As seguintes métricas do CloudWatch são recomendadas para monitorar o desempenho do seu pipeline de ingestão. Essas métricas podem ajudar você a identificar a quantidade de dados processados nas exportações, a quantidade de eventos processados nos fluxos, os erros no processamento de exportações e eventos de fluxo e o número de documentos gravados no destino. Você pode configurar alarmes do CloudWatch para executar uma ação quando uma dessas métricas excede um valor especificado por um período especificado.