Utilisation d'un pipeline OpenSearch d'ingestion avec Amazon DynamoDB

Vous pouvez utiliser un pipeline d' OpenSearch ingestion avec DynamoDB pour diffuser des événements de table DynamoDB (tels que la création, la mise à jour et la suppression) vers les domaines et les collections Amazon Service. OpenSearch Le pipeline OpenSearch d'ingestion intègre une infrastructure de capture des données de modification (CDC) afin de fournir un moyen à grande échelle et à faible latence de diffuser en continu des données à partir d'une table DynamoDB.

Vous pouvez utiliser DynamoDB comme source pour traiter les données de deux manières : avec ou sans capture initiale complète.

Un instantané initial complet est une sauvegarde d'une table réalisée par DynamoDB avec point-in-time la fonctionnalité recovery PITR (). DynamoDB télécharge cet instantané sur Amazon S3. À partir de là, un pipeline d' OpenSearch ingestion l'envoie vers un index d'un domaine ou le partitionne en plusieurs index d'un domaine. Pour garantir la cohérence des données dans OpenSearch DynamoDB, le pipeline synchronise tous les événements de création, de mise à jour et de suppression de la table DynamoDB avec les documents enregistrés dans l'index ou les index. OpenSearch

Lorsque vous utilisez un instantané initial complet, votre pipeline d' OpenSearch ingestion ingère d'abord l'instantané, puis commence à lire les données provenant de DynamoDB Streams. Il finit par rattraper le retard et maintenir la cohérence des données en temps quasi réel entre DynamoDB et. OpenSearch Lorsque vous choisissez cette option, vous devez activer les deux options PITR ainsi qu'un flux DynamoDB sur votre table.

Vous pouvez également utiliser l'intégration d' OpenSearch ingestion avec DynamoDB pour diffuser des événements sans capture instantanée. Choisissez cette option si vous disposez déjà d'un instantané complet d'un autre mécanisme, ou si vous souhaitez simplement diffuser les événements actuels à partir d'une table DynamoDB avec DynamoDB Streams. Lorsque vous choisissez cette option, il vous suffit d'activer un flux DynamoDB sur votre table.

Pour plus d'informations sur cette intégration, consultez la section ETLIntégration zéro de DynamoDB à OpenSearch Amazon Service dans Amazon DynamoDB le Guide du développeur.

Prérequis

Pour configurer votre pipeline, vous devez disposer d'une table DynamoDB sur laquelle DynamoDB Streams est activé. Votre stream doit utiliser le type de NEW_IMAGE stream view. Cependant, les pipelines OpenSearch d'ingestion peuvent également diffuser des événements NEW_AND_OLD_IMAGES si ce type de vue de flux correspond à votre cas d'utilisation.

Si vous utilisez des instantanés, vous devez également activer la point-in-time restauration sur votre table. Pour plus d'informations, consultez les sections Création d'une table, point-in-time Activation de la restauration et Activation d'un flux dans le guide du développeur Amazon DynamoDB.

Étape 1 : configurer le rôle du pipeline

Une fois votre table DynamoDB configurée, configurez le rôle de pipeline que vous souhaitez utiliser dans la configuration de votre pipeline et ajoutez les autorisations DynamoDB suivantes dans le rôle :

{
    "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}/*"
            ]
        }
    ]
}

Vous pouvez également utiliser une clé gérée par AWS KMS le client pour chiffrer les fichiers de données d'exportation. Pour déchiffrer les objets exportés, spécifiez s3_sse_kms_key_id l'ID de clé dans la configuration d'exportation du pipeline au format suivant :arn:aws:kms:us-west-2:{account-id}:key/my-key-id. La politique suivante inclut les autorisations requises pour utiliser une clé gérée par le client :

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

Étape 2 : Création du pipeline

Vous pouvez ensuite configurer un pipeline d' OpenSearch ingestion comme le suivant, qui spécifie DynamoDB comme source. Cet exemple de pipeline ingère des données provenant de table-a l'PITRinstantané, suivies d'événements provenant de DynamoDB Streams. Une position de départ de LATEST indique que le pipeline doit lire les dernières données de 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"

Vous pouvez utiliser un plan DynamoDB préconfiguré pour créer ce pipeline. Pour de plus amples informations, veuillez consulter Utiliser des plans pour créer un pipeline.

Cohérence des données

OpenSearch L'ingestion prend en charge end-to-end la reconnaissance afin de garantir la durabilité des données. Lorsqu'un pipeline lit des instantanés ou des flux, il crée dynamiquement des partitions pour un traitement parallèle. Le pipeline marque une partition comme terminée lorsqu'il reçoit un accusé de réception après avoir ingéré tous les enregistrements du OpenSearch domaine ou de la collection.

Si vous souhaitez intégrer des données dans une collection de recherche OpenSearch sans serveur, vous pouvez générer un identifiant de document dans le pipeline. Si vous souhaitez intégrer des données dans une collection de séries chronologiques OpenSearch sans serveur, notez que le pipeline ne génère pas d'identifiant de document.

Un pipeline d' OpenSearch ingestion fait également correspondre les actions des événements entrants aux actions d'indexation groupées correspondantes pour faciliter l'ingestion de documents. Cela garantit la cohérence des données, de sorte que chaque modification de données dans DynamoDB est conciliée avec les modifications de document correspondantes. OpenSearch

Types de données de mappage

OpenSearch Le service fait correspondre dynamiquement les types de données de chaque document entrant au type de données correspondant dans DynamoDB. Le tableau suivant montre comment OpenSearch Service mappe automatiquement les différents types de données.

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

Nous vous recommandons de configurer la file d'attente des lettres mortes (DLQ) dans votre pipeline d' OpenSearch ingestion. Si vous avez configuré la file d'attente, le OpenSearch service envoie tous les documents défaillants qui ne peuvent pas être ingérés en raison d'échecs de mappage dynamique vers la file d'attente.

En cas d'échec des mappages automatiques, vous pouvez utiliser template_type et template_content dans la configuration de votre pipeline pour définir des règles de mappage explicites. Vous pouvez également créer des modèles de mappage directement dans votre domaine de recherche ou votre collection avant de démarrer le pipeline.

Limites

Tenez compte des limites suivantes lorsque vous configurez un pipeline d' OpenSearch ingestion pour DynamoDB :

CloudWatch Alarmes recommandées pour DynamoDB

Les CloudWatch mesures suivantes sont recommandées pour surveiller les performances de votre pipeline d'ingestion. Ces indicateurs peuvent vous aider à identifier la quantité de données traitées à partir des exportations, le nombre d'événements traités à partir des flux, les erreurs lors du traitement des exportations et des événements des flux, ainsi que le nombre de documents écrits vers la destination. Vous pouvez configurer des CloudWatch alarmes pour effectuer une action lorsque l'une de ces mesures dépasse une valeur spécifiée pendant une durée spécifiée.