Uso de una canalización de OpenSearch Ingestion con Amazon DynamoDB

Puede usar el complemento DynamoDB para transmitir eventos de tablas, como creaciones, actualizaciones y eliminaciones, a los dominios de Amazon OpenSearch Service y a las colecciones de Amazon OpenSearch sin servidor. La canalización utiliza la captura de datos de cambios (CDC) para la transmisión de alta escala y baja latencia.

Puede procesar datos de DynamoDB con o sin una instantánea inicial completa.

Para obtener más información, consulte Integración sin ETL de DynamoDB con Amazon OpenSearch Service en la Guía para desarrolladores de Amazon DynamoDB.

Requisitos previos

Para configurar la canalización, debe tener una tabla de DynamoDB con DynamoDB Streams activado. La transmisión debe usar el tipo de vista de transmisión NEW_IMAGE. Sin embargo, las canalizaciones de OpenSearch Ingestion también pueden transmitir eventos con NEW_AND_OLD_IMAGES si este tipo de vista de transmisión se ajusta a su caso de uso.

Si utiliza instantáneas, también debe habilitar la recuperación en un momento dado en su tabla. Para obtener más información, consulte Creación de una tabla, Habilitación de la recuperación en un momento dado y Habilitación de una transmisión en la Guía para desarrolladores de Amazon DynamoDB.

Paso 1: configurar el rol de canalización

Una vez configurada la tabla de DynamoDB, configure el rol de canalización que desee usar en la configuración de canalización y añada los siguientes permisos de DynamoDB al rol:

También puede usar una clave administrada por el cliente de AWS KMS para cifrar los archivos de datos de exportación. Para descifrar los objetos exportados, especifique s3_sse_kms_key_id para el ID de clave en la configuración de exportación de la canalización con el siguiente formato: arn:aws:kms:region:account-id:key/my-key-id. La siguiente política incluye los permisos necesarios para utilizar una clave administrada por el cliente:

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

Paso 2: crear la canalización

A continuación, puede configurar una canalización de OpenSearch Ingestion como la siguiente, en la que se especifica a DynamoDB como el origen. Este ejemplo de canalización incorpora los datos de la table-a con la instantánea de PITR, seguidos de los eventos de DynamoDB Streams. Una posición inicial de LATEST indica que la canalización debe leer los datos más recientes de 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"

Puede utilizar un esquema preconfigurado de DynamoDB para crear esta canalización. Para obtener más información, consulte Uso de esquemas.

Coherencia de datos

OpenSearch Ingestion admite el reconocimiento integral para garantizar la durabilidad de los datos. Cuando una canalización lee instantáneas o transmisiones, crea particiones de forma dinámica para el procesamiento paralelo. La canalización marca una partición como completa cuando recibe un reconocimiento después de incorporar todos los registros en el dominio o la colección de OpenSearch.

Si desea incorporarlos a una colección de búsquedas de OpenSearch sin servidor, puede generar un ID de documento en la canalización. Si quiere incorporarlos a una colección de series temporales de OpenSearch sin servidor, tenga en cuenta que la canalización no genera un ID de documento.

Una canalización de OpenSearch Ingestion también asigna las acciones de los eventos entrantes a las correspondientes acciones de indexación masiva para facilitar la incorporación de documentos. Esto mantiene la coherencia de datos, de modo que cada cambio de datos en DynamoDB se concilie con los cambios de documento correspondientes en OpenSearch.

Asignación de tipos de datos

OpenSearch Service asigna de manera dinámica los tipos de datos de cada documento entrante al tipo de datos correspondiente en DynamoDB. En la siguiente tabla se muestra cómo OpenSearch Service asigna de forma automática varios tipos de datos.

{
 "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"
      }
     }
    }
   }
  }
 }
}

Le recomendamos configurar la cola de mensajes fallidos (DLQ) en su canalización de OpenSearch Ingestion. Si ha configurado la cola, OpenSearch Service envía todos los documentos con errores que no se pueden incorporar debido a errores de asignación dinámica.

En caso de que las asignaciones automáticos fallen, puede usar template_type y template_content en su configuración de canalización para definir reglas de asignación explícitas. Como alternativa, puede crear plantillas de asignación directamente en su dominio o colección de búsqueda antes de iniciar la canalización.

Limitaciones

Tenga en cuenta las siguientes limitaciones cuando configure una canalización de OpenSearch Ingestion para DynamoDB:

Alarmas de CloudWatch recomendadas para DynamoDB

Se recomiendan las siguientes métricas de CloudWatch para supervisar el rendimiento de la canalización de ingesta. Estas métricas pueden ayudarle a identificar la cantidad de datos procesados a partir de exportaciones y flujos, los errores al procesar eventos de transmisión y exportación y la cantidad de documentos escritos en el destino. Puede configurar las alarmas de CloudWatch para que lleven a cabo una acción cuando una de estas métricas supere un valor especificado durante un periodo de tiempo determinado.