将 OpenSearch Ingestion 管道与 Amazon DynamoDB 结合使用

您可以使用 DynamoDB 插件将表事件（例如创建、更新和删除）流式传输到 Amazon OpenSearch Service 域和 Amazon OpenSearch 无服务器集合。该管道采用变更数据捕获（CDC）技术实现大规模、低延迟的流式传输。

无论有没有完整初始快照，都可以处理 DynamoDB 数据。

有关的更多信息，请参阅《Amazon DynamoDB 开发人员指南》中的 DynamoDB zero-ETL integration with Amazon OpenSearch Service。

先决条件

要设置管道，您必须有一个已启用 DynamoDB Streams 的 DynamoDB 表。您的流应使用 NEW_IMAGE 流视图类型。但是，如果这种流视图类型适合您的用例，OpenSearch Ingestion 管道也可以使用 NEW_AND_OLD_IMAGES 流式传输事件。

如果使用快照，则还必须启用表中的时间点恢复。有关更多信息，请参阅《Amazon DynamoDB 开发人员指南》中的创建表、启用时间点恢复和启用流。

步骤 1：配置管道角色

设置 DynamoDB 表后，设置要在管道配置中使用的管道角色，并在该角色中添加以下 DynamoDB 权限：

您也可以使用 Amazon KMS 客户托管密钥加密导出数据文件。要解密导出的对象，请在管道的导出配置中指定 s3_sse_kms_key_id 作为密钥 ID，格式如下：arn:aws:kms:region:account-id:key/my-key-id。以下策略包括使用客户自主管理型密钥时的必需权限：

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

步骤 2：创建管道

然后，您可以配置如下所示的 OpenSearch Ingestion 管道，指定 DynamoDB 作为源。此示例管道使用 PITR 快照从 table-a 中摄取数据，然后从 DynamoDB Streams 摄取事件。LATEST 的起始位置指示管道应从 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"

您可以使用预先配置的 DynamoDB 蓝图，以创建此管道。有关更多信息，请参阅 使用蓝图。

数据一致性

OpenSearch Ingestion 支持端到端确认以确保数据持久性。管道读取快照或流时，它会动态创建分区以进行并行处理。管道在摄取 OpenSearch 域或集合中的所有记录后收到确认时，会将分区标记为已完成。

如果要摄取到 OpenSearch 无服务器搜索集合中，您可以在管道中生成文档 ID。如果要摄取到 OpenSearch 无服务器时间序列集合中，请注意管道不会生成文档 ID。

OpenSearch Ingestion 管道还会将传入事件操作映射到相应的批量索引操作，以帮助摄取文档。这可保持数据一致性，以使 DynamoDB 中的每个数据更改都与 OpenSearch 中的相应文档更改保持一致。

映射数据类型

OpenSearch Service 将每个传入文档中的数据类型动态映射到 DynamoDB 中相应的数据类型。下表显示 OpenSearch Service 如何自动映射各种数据类型。

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

我们建议在 OpenSearch Ingestion 管道中配置死信队列（DLQ）。如果已配置该队列，OpenSearch Service 会将所有因动态映射失败而无法摄取的失败文档发送到该队列。

如果自动映射失败，则可以在管道配置中使用 template_type 和 template_content 来定义显式映射规则。或者，您可以在启动管道之前直接在搜索域或集合中创建映射模板。

限制

为 DynamoDB 设置 OpenSearch Ingestion 管道时，请考虑以下限制：

适用于 DynamoDB 的建议 CloudWatch 警报

建议使用以下 CloudWatch 指标来监控摄取管道的性能。这些指标可能有助您确定处理的导出数据量、处理的流事件量、处理导出和流事件时的错误数以及写入目标的文档数量。您可以设置 CloudWatch 警报，从而在其中任何指标在指定期限内超出指定的值时执行某个操作。