Usando um pipeline de OpenSearch ingestão com o Amazon DynamoDB

Você pode usar um pipeline de OpenSearch ingestão com o DynamoDB para transmitir eventos de tabela do DynamoDB (como criar, atualizar e excluir) para domínios e coleções do Amazon Service. OpenSearch O pipeline OpenSearch de ingestão incorpora a infraestrutura de captura de dados de alteração (CDC) para fornecer uma forma de alta escala e baixa latência de transmitir dados continuamente de uma tabela do DynamoDB.

Há duas maneiras de usar o DynamoDB como origem para processar dados: com e sem um snapshot inicial completo.

Um snapshot inicial completo é um backup de uma tabela que o DynamoDB faz com point-in-time o recurso recovery (). PITR O DynamoDB carrega esse snapshot no Amazon S3. A partir daí, um pipeline de OpenSearch ingestão o envia para um índice em um domínio ou o particiona em vários índices em um domínio. Para manter os dados no DynamoDB OpenSearch e consistentes, o pipeline sincroniza todos os eventos de criação, atualização e exclusão na tabela do DynamoDB com os documentos salvos no índice ou índices. OpenSearch

Quando você usa um snapshot inicial completo, seu pipeline de OpenSearch ingestão primeiro ingere o snapshot e depois começa a ler os dados do DynamoDB Streams. Eventualmente, ele recupera e mantém a consistência de dados quase em tempo real entre o DynamoDB e. OpenSearch Ao escolher essa opção, você deve habilitar ambos PITR e um stream do DynamoDB na sua tabela.

Você também pode usar a integração do OpenSearch Ingestion com o DynamoDB para transmitir eventos sem um snapshot. Escolha essa opção se você já tiver um snapshot completo de algum outro mecanismo ou se quiser apenas transmitir eventos atuais de uma tabela do DynamoDB com o DynamoDB Streams. Ao escolher essa opção, você só precisa habilitar um fluxo do DynamoDB na sua tabela.

Para obter mais informações sobre essa integração, consulte ETLIntegração zero do DynamoDB com o OpenSearch Amazon Service no Amazon DynamoDB Guia do desenvolvedor.

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. No entanto, os pipelines de OpenSearch ingestão também podem transmitir eventos NEW_AND_OLD_IMAGES se esse tipo de visualização de fluxo for adequado ao seu caso de uso.

Se você estiver usando instantâneos, também deverá ativar a point-in-time recuperação em sua tabela. Para obter mais informações, consulte Criar uma tabela, Habilitar a point-in-time recuperação e Habilitar um stream no Amazon DynamoDB Developer Guide.

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:

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "allowRunExportJob",
            "Effect": "Allow",
            "Action": [
                "dynamodb:DescribeTable",
                "dynamodb:DescribeContinuousBackups",
                "dynamodb:ExportTableToPointInTime"
            ],
            "Resource": [
                "arn:aws:dynamodb:us-east-1:{account-id}:table/my-table"
            ]
        },
        {
            "Sid": "allowCheckExportjob",
            "Effect": "Allow",
            "Action": [
                "dynamodb:DescribeExport"
            ],
            "Resource": [
                "arn:aws:dynamodb:us-east-1:{account-id}:table/my-table/export/*"
            ]
        },
        {
            "Sid": "allowReadFromStream",
            "Effect": "Allow",
            "Action": [
                "dynamodb:DescribeStream",
                "dynamodb:GetRecords",
                "dynamodb:GetShardIterator"
            ],
            "Resource": [
                "arn:aws:dynamodb:us-east-1:{account-id}:table/my-table/stream/*"
            ]
        },
        {
            "Sid": "allowReadAndWriteToS3ForExport",
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:AbortMultipartUpload",
                "s3:PutObject",
                "s3:PutObjectAcl"
            ],
            "Resource": [
                "arn:aws:s3:::my-bucket/{exportPath}/*"
            ]
        }
    ]
}

Você também pode usar uma chave gerenciada pelo AWS KMS cliente 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:us-west-2:{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:us-west-2:{account-id}:key/my-key-id
}

Etapa 2: Criar o pipeline

Em seguida, você pode configurar um pipeline OpenSearch de ingestão como o seguinte, que especifica o DynamoDB como origem. Esse exemplo de pipeline ingere dados table-a com o PITR snapshot, seguidos 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:us-west-2:{account-id}:table/table-a"  
        export:
          s3_bucket: "my-bucket"
          s3_prefix: "export/"
        stream:
          start_position: "LATEST"
      aws:
        region: "us-west-2"
        sts_role_arn: "arn:aws:iam::{account-id}:role/pipeline-role"
  sink:
  - opensearch:
      hosts: ["https://search-mydomain.us-east-1.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 blueprint pré-configurado do DynamoDB para criar esse pipeline. Para obter mais informações, consulte Usar esquemas para criar um pipeline.

Consistência de dados

OpenSearch A ingestão suporta o end-to-end reconhecimento 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 ela recebe uma confirmação após a ingestão de todos os registros no OpenSearch domínio ou na coleção.

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

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

Mapeaar tipo de dados

OpenSearch O serviço mapeia dinamicamente os tipos de dados 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 letras mortas (DLQ) em seu OpenSearch pipeline de ingestão. Se você configurou a fila, o OpenSearch Service envia todos os documentos com falha que não podem 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 de OpenSearch ingestão para o DynamoDB:

CloudWatch Alarmes recomendados para o DynamoDB

As CloudWatch métricas a seguir são recomendadas para monitorar o desempenho do seu pipeline de ingestão. Essas métricas podem ajudá-lo 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 CloudWatch alarmes para realizar uma ação quando uma dessas métricas exceder um valor especificado por um determinado período de tempo.