Verwenden einer OpenSearch Ingestion-Pipeline mit Amazon DynamoDB

Sie können eine OpenSearch Ingestion-Pipeline mit DynamoDB verwenden, um DynamoDB-Tabellenereignisse (wie Erstellen, Aktualisieren und Löschen) in Amazon Service-Domänen und Sammlungen zu streamen. OpenSearch Die OpenSearch Ingestion-Pipeline umfasst eine Change Data Capture (CDC) -Infrastruktur, um eine hochskalierbare Methode mit geringer Latenz für das kontinuierliche Streamen von Daten aus einer DynamoDB-Tabelle bereitzustellen.

Es gibt zwei Möglichkeiten, DynamoDB als Quelle für die Datenverarbeitung zu verwenden — mit und ohne einen vollständigen anfänglichen Snapshot.

Ein vollständiger Anfangssnapshot ist eine Sicherung einer Tabelle, die DynamoDB mit der Funktion point-in-time recovery (PITR) erstellt. DynamoDB lädt diesen Snapshot auf Amazon S3 hoch. Von dort aus sendet eine OpenSearch Ingestion-Pipeline ihn an einen Index in einer Domain oder partitioniert ihn in mehrere Indizes in einer Domain. Damit die Daten in DynamoDB OpenSearch konsistent bleiben, synchronisiert die Pipeline alle Erstellungs-, Aktualisierungs- und Löschereignisse in der DynamoDB-Tabelle mit den Dokumenten, die im Index oder den Indizes gespeichert sind. OpenSearch

Wenn Sie einen vollständigen Anfangssnapshot verwenden, nimmt Ihre OpenSearch Ingestion-Pipeline zuerst den Snapshot auf und beginnt dann, Daten aus DynamoDB Streams zu lesen. Es holt schließlich auf und gewährleistet nahezu in Echtzeit die Datenkonsistenz zwischen DynamoDB und. OpenSearch Wenn Sie diese Option wählen, müssen Sie PITR sowohl als auch einen DynamoDB-Stream in Ihrer Tabelle aktivieren.

Sie können auch die OpenSearch Ingestion-Integration mit DynamoDB verwenden, um Ereignisse ohne Snapshot zu streamen. Wählen Sie diese Option, wenn Sie bereits einen vollständigen Snapshot von einem anderen Mechanismus haben oder wenn Sie nur aktuelle Ereignisse aus einer DynamoDB-Tabelle mit DynamoDB Streams streamen möchten. Wenn Sie diese Option wählen, müssen Sie nur einen DynamoDB-Stream in Ihrer Tabelle aktivieren.

Weitere Informationen zu dieser Integration finden Sie unter DynamoDB zero- ETL integration with Amazon OpenSearch Service im Amazon DynamoDB Developer Guide.

Voraussetzungen

Um Ihre Pipeline einzurichten, benötigen Sie eine DynamoDB-Tabelle mit aktivierten DynamoDB Streams. Ihr Stream sollte den Stream-View-Typ verwenden. NEW_IMAGE OpenSearch Ingestion-Pipelines können jedoch auch Ereignisse streamen, NEW_AND_OLD_IMAGES sofern dieser Stream-View-Typ zu Ihrem Anwendungsfall passt.

Wenn Sie Snapshots verwenden, müssen Sie auch die point-in-time Wiederherstellung für Ihre Tabelle aktivieren. Weitere Informationen finden Sie unter Erstellen einer Tabelle, Aktivieren der point-in-time Wiederherstellung und Aktivieren eines Streams im Amazon DynamoDB DynamoDB-Entwicklerhandbuch.

Schritt 1: Konfigurieren Sie die Pipeline-Rolle

Nachdem Sie Ihre DynamoDB-Tabelle eingerichtet haben, richten Sie die Pipeline-Rolle ein, die Sie in Ihrer Pipeline-Konfiguration verwenden möchten, und fügen Sie der Rolle die folgenden DynamoDB-Berechtigungen hinzu:

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

Sie können auch einen vom AWS KMS Kunden verwalteten Schlüssel verwenden, um die Exportdatendateien zu verschlüsseln. Um die exportierten Objekte zu entschlüsseln, geben Sie s3_sse_kms_key_id für die Schlüssel-ID in der Exportkonfiguration der Pipeline das folgende Format an:. arn:aws:kms:us-west-2:{account-id}:key/my-key-id Die folgende Richtlinie umfasst die erforderlichen Berechtigungen für die Verwendung eines vom Kunden verwalteten Schlüssels:

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

Schritt 2: Erstellen Sie die Pipeline

Anschließend können Sie eine OpenSearch Ingestion-Pipeline wie die folgende konfigurieren, die DynamoDB als Quelle angibt. Diese Beispielpipeline nimmt Daten aus table-a dem PITR Snapshot auf, gefolgt von Ereignissen aus DynamoDB Streams. Die Startposition von LATEST gibt an, dass die Pipeline die neuesten Daten aus DynamoDB Streams lesen soll.

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"

Sie können einen vorkonfigurierten DynamoDB-Blueprint verwenden, um diese Pipeline zu erstellen. Weitere Informationen finden Sie unter Verwenden von Blueprints zum Erstellen einer Pipeline.

Datenkonsistenz

OpenSearch Die Datenaufnahme unterstützt end-to-end die Bestätigung, um die Datenbeständigkeit sicherzustellen. Wenn eine Pipeline Snapshots oder Streams liest, erstellt sie dynamisch Partitionen für die Parallelverarbeitung. Die Pipeline markiert eine Partition als abgeschlossen, wenn sie nach der Aufnahme aller Datensätze in der OpenSearch Domäne oder Sammlung eine Bestätigung erhält.

Wenn Sie Daten in eine OpenSearch serverlose Suchsammlung aufnehmen möchten, können Sie in der Pipeline eine Dokument-ID generieren. Wenn Sie Daten in eine OpenSearch serverlose Zeitreihensammlung aufnehmen möchten, beachten Sie, dass die Pipeline keine Dokument-ID generiert.

Eine OpenSearch Ingestion-Pipeline ordnet außerdem eingehende Ereignisaktionen den entsprechenden Massenindizierungsaktionen zu, um das Ingestieren von Dokumenten zu erleichtern. Dadurch bleiben die Daten konsistent, sodass jede Datenänderung in DynamoDB mit den entsprechenden Dokumentänderungen in abgeglichen wird. OpenSearch

Datentypen zuordnen

OpenSearch Der Service ordnet Datentypen in jedem eingehenden Dokument dynamisch dem entsprechenden Datentyp in DynamoDB zu. Die folgende Tabelle zeigt, wie OpenSearch Service verschiedene Datentypen automatisch zuordnet.

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

Wir empfehlen, dass Sie die Warteschlange für unzustellbare Briefe (DLQ) in Ihrer Ingestion-Pipeline konfigurieren. OpenSearch Wenn Sie die Warteschlange konfiguriert haben, sendet OpenSearch Service alle fehlgeschlagenen Dokumente, die aufgrund von Fehlern bei der dynamischen Zuordnung nicht aufgenommen werden konnten, zur Warteschlange.

Falls automatische Zuordnungen fehlschlagen, können Sie template_type und template_content in Ihrer Pipeline-Konfiguration verwenden, um explizite Zuordnungsregeln zu definieren. Alternativ können Sie Zuordnungsvorlagen direkt in Ihrer Suchdomain oder Sammlung erstellen, bevor Sie die Pipeline starten.

Einschränkungen

Beachten Sie die folgenden Einschränkungen, wenn Sie eine OpenSearch Ingestion-Pipeline für DynamoDB einrichten:

Empfohlene CloudWatch Alarme für DynamoDB

Die folgenden CloudWatch Metriken werden für die Überwachung der Leistung Ihrer Datenerfassungspipeline empfohlen. Mithilfe dieser Metriken können Sie die Menge der im Rahmen von Exporten verarbeiteten Daten, die Anzahl der aus Streams verarbeiteten Ereignisse, die Fehler bei der Verarbeitung von Exporten und Stream-Ereignissen sowie die Anzahl der Dokumente, die an das Ziel geschrieben wurden, ermitteln. Sie können CloudWatch Alarme einrichten, um eine Aktion auszuführen, wenn eine dieser Metriken für einen bestimmten Zeitraum einen bestimmten Wert überschreitet.