在 Amazon DocumentDB 中使用 OpenSearch 摄取管道

聚焦模式

在 Amazon DocumentDB 中使用 OpenSearch 摄取管道 - 亚马逊 OpenSearch 服务

先决条件步骤 1：配置管道角色步骤 2：创建管道数据一致性映射数据类型限制推荐的 CloudWatch 警报

您可以使用 D ocumentDB 插件将文档更改（例如创建、更新和删除）流式传输到亚马逊 OpenSearch 服务。该管道支持变更数据捕获 (CDC)（如果有），或者支持高规模、低延迟流媒体的 API 轮询。

无论有没有完整的初始快照，您都可以处理数据。完整快照会捕获整个 Amazon DocumentDB 馆藏并将其上传到 Amazon S3。然后，管道将数据发送到一个或多个 OpenSearch 索引。在摄取快照后，管道会同步正在进行的更改以保持一致性，并最终赶上近乎实时的更新。

如果您已经有来自其他来源的完整快照，或者只需要处理新事件，则可以在没有快照的情况下进行流式传输。在这种情况下，管道直接从 Amazon DocumentDB 变更流中读取，无需初始批量加载。

如果您启用直播，则必须在 Amazon DocumentDB 馆藏上启用更改流。但是，如果您只执行满载或导出，则不需要更改流。

先决条件

在创建 OpenSearch 摄取管道之前，请执行以下步骤：

按照《Amazon DocumentDB 开发者指南》中 Create an Amazon DocumentDB cluster 说明的步骤操作，创建一个具有数据读取权限的 Amazon DocumentDB 集群。如果您使用 CDC 基础设施，请将您的 Amazon DocumentDB 集群配置为发布变更流。
在 Amazon DocumentDB 集群上启用 TLS。
设置私有地址空间的 VPC CIDR 以用于 OpenSearch 接入。
使用在您的亚马逊文档数据库集群上设置身份验证。 AWS Secrets Manager按照 Automatically Rotating Passwords for Amazon DocumentDB 中的步骤启用密钥轮换。有关更多信息，请参阅 Database Access Using Role-Based Access Control 和 Security in Amazon DocumentDB。
如果您使用更改流订阅 Amazon DocumentDB 集合中的数据更改，请使用 change_stream_log_retention_duration 参数将保留期延长至最多 7 天，以避免数据丢失。默认情况下，更改流事件将在事件录制后存储 3 小时，这对于大型集合来说是不够的。要修改更改流的保留期，请参阅 Modifying the Change Stream Log Retention Duration。
创建 OpenSearch 服务域或 OpenSearch 无服务器集合。有关更多信息，请参阅创建 OpenSearch 服务域和创建集合。
将基于资源的策略附加到域，或将数据访问策略附加到集合。这些访问策略允许 OpenSearch Ingestion 将数据从您的 Amazon DocumentDB 集群写入您的域名或集合。

以下示例域访问策略允许您在下一步中创建的管道角色将数据写入域。确保使用自身 ARN 更新 resource。
```
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::pipeline-account-id:role/pipeline-role"
      },
      "Action": [
        "es:DescribeDomain",
        "es:ESHttp*"
      ],
      "Resource": [
        "arn:aws:es:region:account-id:domain/domain-name"
      ]
    }
  ]
}
```
要创建具有正确权限的 IAM 角色来访问向集合或域写入数据，请参阅在 Amazon OpenSearch Ingestion 中设置角色和用户。

步骤 1：配置管道角色

完成 Amazon DocumentDB 管道先决条件设置后，配置管道角色以在管道配置中使用，并为该角色添加以下 Amazon DocumentDB 权限：


{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "allowS3ListObjectAccess",
            "Effect": "Allow",
            "Action": [
                "s3:ListBucket"
            ],
            "Resource": [
                "arn:aws:s3:::s3-bucket"
            ],
            "Condition": {
                "StringLike": {
                    "s3:prefix": "s3-prefix/*"
                }
            }
        },
        {
            "Sid": "allowReadAndWriteToS3ForExportStream",
            "Effect": "Allow",
            "Action": [
                "s3:PutObject",
                "s3:GetObject",
                "s3:DeleteObject"
            ],
            "Resource": [
                "arn:aws:s3:::s3-bucket/s3-prefix/*"
            ]
        },
        {
            "Sid": "SecretsManagerReadAccess",
            "Effect": "Allow",
            "Action": [
                "secretsmanager:GetSecretValue"
            ],
            "Resource": ["arn:aws:secretsmanager:region:account-id:secret:secret-name"]
        },
        {
            "Effect": "Allow",
            "Action": [
                "ec2:AttachNetworkInterface",
                "ec2:CreateNetworkInterface",
                "ec2:CreateNetworkInterfacePermission",
                "ec2:DeleteNetworkInterface",
                "ec2:DeleteNetworkInterfacePermission",
                "ec2:DetachNetworkInterface",
                "ec2:DescribeNetworkInterfaces"
            ],
            "Resource": [
                "arn:aws:ec2:*:account-id:network-interface/*",
                "arn:aws:ec2:*:account-id:subnet/*",
                "arn:aws:ec2:*:account-id:security-group/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "ec2:DescribeDhcpOptions",
                "ec2:DescribeRouteTables",
                "ec2:DescribeSecurityGroups",
                "ec2:DescribeSubnets",
                "ec2:DescribeVpcs",
                "ec2:Describe*"
            ],
            "Resource": "*"
        },
        { 
            "Effect": "Allow",
            "Action": [ 
                "ec2:CreateTags"
            ],
            "Resource": "arn:aws:ec2:*:*:network-interface/*",
            "Condition": { 
               "StringEquals": 
                    {
                        "aws:RequestTag/OSISManaged": "true"
                    } 
            } 
        }
    ]
}

您必须为用于创建 OpenSearch Ingestion 管道的 IAM 角色提供上述 Amazon EC2 权限，因为管道使用这些权限在您的 VPC 中创建和删除网络接口。该管道只能通过此网络接口访问 Amazon DocumentDB 集群。

步骤 2：创建管道

然后，您可以配置如下所示的 OpenSearch 摄取管道，该管道将 Amazon DocumentDB 指定为来源。请注意，为了填充索引名称，getMetadata 函数会将 documentdb_collection 用作元数据键。要在不使用 getMetadata 方法的情况下使用其他索引名称，则可以使用配置 index: "my_index_name"。


version: "2"
documentdb-pipeline:
  source:
    documentdb:
      acknowledgments: true
      host: "https://docdb-cluster-id.us-east-1.docdb.amazonaws.com"
      port: 27017
      authentication:
        username: ${aws_secrets:secret:username}
        password: ${aws_secrets:secret:password}
      aws:
        sts_role_arn: "arn:aws:iam::account-id:role/pipeline-role"
      s3_bucket: "bucket-name"
      s3_region: "bucket-region" 
      s3_prefix: "path" #optional path for storing the temporary data
      collections:
        - collection: "dbname.collection"
          export: true
          stream: true
  sink:
  - opensearch:
      hosts: ["https://search-mydomain.us-east-1.es.amazonaws.com"]
      index: "${getMetadata(\"documentdb_collection\")}"
      index_type: custom
      document_id: "${getMetadata(\"primary_key\")}"
      action: "${getMetadata(\"opensearch_action\")}"
      document_version: "${getMetadata(\"document_version\")}"
      document_version_type: "external"
extension:
  aws:
    secrets:
      secret:
        secret_id: "my-docdb-secret"
        region: "us-east-1"
        sts_role_arn: "arn:aws:iam::account-id:role/pipeline-role"
        refresh_interval: PT1H

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

如果您使用创建管道，则还必须将管道连接到 VPC，才能使用 Amazon DocumentDB 作为来源。 AWS Management Console 要执行此操作，请找到网络配置部分，选择连接到 VPC 复选框，然后从提供的任意一个默认选项中选择 CIDR，或者选择自己的 CIDR。您可以按照 RFC 1918 当前最佳实践中的定义，使用私有地址空间中的任何 CIDR。

要提供自定义 CIDR，请从下拉菜单中选择其他。为避免 OpenSearch Ingestion 和 Amazon DocumentDB 之间的 IP 地址冲突，请确保亚马逊 DocumentDB VPC CIDR 与用于摄取的 CIDR 不同。 OpenSearch

有关更多信息，请参阅为管道配置 VPC 访问权限。

数据一致性

该管道通过持续轮询或接收来自 Amazon DocumentDB 集群的更改以及更新索引中的相应文档来确保数据一致性。 OpenSearch

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

如果要收录到 OpenSearch 无服务器搜索集合中，可以在管道中生成文档 ID。如果要采集到 OpenSearch Serverless 时间序列集合，请注意管道不会生成文档 ID，因此您必须在工作流接收器配置document_id: "${getMetadata(\"primary_key\")}"中省略文档 ID。

In OpenSearch gestion 管道还将传入的事件操作映射到相应的批量索引操作中，以帮助采集文档。这样可以保持数据的一致性，从而使 Amazon DocumentDB 中的每一次数据更改都与中的相应文档更改保持一致。 OpenSearch

映射数据类型

OpenSearch 服务将每个传入文档中的数据类型动态映射到 Amazon DocumentDB 中的相应数据类型。下表显示了 S OpenSearch ervice 如何自动映射各种数据类型。

数据类型	OpenSearch	Amazon DocumentDB
整数	OpenSearch 自动将 Amazon DocumentDB 的整数值映射到 OpenSearch整数。 OpenSearch 根据第一个发送的文档动态映射该字段。如果 Amazon DocumentDB 中的同一属性混合了多种数据类型，则自动映射可能会失败。例如，如果您的第一个文档的属性很长，而后来的文档具有与整数相同的属性， OpenSearch 则无法收录第二个文档。对于这些情况，应提供选择最灵活数值类型的显式映射模板，如下所示： `{ "template": { "mappings": { "properties": { "MixedNumberField": { "type": "float" } } } } }`	Amazon DocumentDB 支持整型。
长整型	OpenSearch 自动将 Amazon DocumentDB 长整数值映射到 OpenSearch多头。 OpenSearch 根据第一个发送的文档动态映射该字段。如果 Amazon DocumentDB 中的同一属性混合了多种数据类型，则自动映射可能会失败。例如，如果您的第一个文档的属性很长，而后来的文档具有与整数相同的属性， OpenSearch 则无法收录第二个文档。对于这些情况，应提供选择最灵活数值类型的显式映射模板，如下所示： `{ "template": { "mappings": { "properties": { "MixedNumberField": { "type": "float" } } } } }`	Amazon DocumentDB 支持长整型。
字符串	OpenSearch 自动将字符串值映射为文本。在某些情况下（例如枚举值），您可以映射到关键字类型。以下示例说明如何将名为的 Amazon DocumentDB 属性映射`PartType`到关键词。 OpenSearch `{ "template": { "mappings": { "properties": { "PartType": { "type": "keyword" } } } } }`	Amazon DocumentDB 支持字符串。
双精度	OpenSearch 自动将 Amazon DocumentDB 双精度值映射到双精度值。 OpenSearch OpenSearch 根据第一个发送的文档动态映射该字段。如果 Amazon DocumentDB 中的同一属性混合了多种数据类型，则自动映射可能会失败。例如，如果您的第一个文档的属性很长，而后来的文档具有与整数相同的属性， OpenSearch 则无法收录第二个文档。对于这些情况，应提供选择最灵活数值类型的显式映射模板，如下所示： `{ "template": { "mappings": { "properties": { "MixedNumberField": { "type": "float" } } } } }`	Amazon DocumentDB 支持双精度浮点数。
日期	默认情况下，日期映射到中的整数 OpenSearch。您可以定义自定义映射模板以将日期映射到 OpenSearch 日期。 `{ "template": { "mappings": { "properties": { "myDateField": { "type": "date", "format": "epoch_second" } } } } }`	Amazon DocumentDB 支持日期。
Timestamp	默认情况下，时间戳映射到中的整数。 OpenSearch您可以定义自定义映射模板以将日期映射到 OpenSearch日期。 `{ "template": { "mappings": { "properties": { "myTimestampField": { "type": "date", "format": "epoch_second" } } } } }`	Amazon DocumentDB 支持时间戳。
布尔值	OpenSearch 将 Amazon DocumentDB 布尔类型映射为 OpenSearch 布尔类型。	Amazon DocumentDB 支持布尔类型属性。
十进制	OpenSearch 将 Amazon DocumentDB 映射属性映射到嵌套字段。嵌套字段内也适用相同的映射。以下示例将嵌套字段中的字符串映射到中的关键字类型 OpenSearch： `{ "template": { "mappings": { "properties": { "myDecimalField": { "type": "double" } } } } }` 使用此自定义映射，您可以对该字段进行双精度查询和聚合。原始值保留了 OpenSearch文档`_source`属性的全部精度。如果没有此映射，则默认 OpenSearch 使用文本。	Amazon DocumentDB 支持小数。
正则表达式	正则表达式类型会创建嵌套字段。这包括 `<myFieldName>.pattern` 和 `<myFieldName>.options`。	Amazon DocumentDB 支持正则表达式。
二进制数据	OpenSearch 自动将 Amazon DocumentDB 二进制数据映射到 OpenSearch文本。您可以提供映射以将它们写成二进制字段 OpenSearch。以下示例说明如何将名为的 Amazon DocumentDB 字段映射`imageData`到 OpenSearch 二进制字段。 `{ "template": { "mappings": { "properties": { "imageData": { "type": "binary" } } } } }`	Amazon DocumentDB 支持二进制数据字段。
ObjectId	具有 ObjectID 类型的字段会映射到 OpenSearch 文本字段。该值将是 objectId 的字符串表示形式。	Amazon DocumentDB 支持 objectIds。
Null	OpenSearch 可以收录 Amazon DocumentDB 空类型的文档。它将该值作为空值保存在文档中。此类型没有映射，并且此字段未编制索引或不可搜索。如果对空类型使用相同的属性名称，然后更改为其他类型（例如字符串），则会为第一个非空值 OpenSearch 创建动态映射。后续值仍然可以是 Amazon DocumentDB 空值。	Amazon DocumentDB 支持空值类型字段。
未定义	OpenSearch 可以收录 Amazon DocumentDB 未定义类型的文档。它将该值作为空值保存在文档中。此类型没有映射，并且此字段未编制索引或不可搜索。如果对未定义的类型使用相同的字段名称，然后更改为其他类型（例如字符串），则会为第一个未定义的值 OpenSearch 创建动态映射。后续值仍然可以是 Amazon DocumentDB 未定义值。	Amazon DocumentDB 支持未定义类型字段。
MinKey	OpenSearch 可以收录亚马逊 DocumentDB minKey 类型的文档。它将该值作为空值保存在文档中。此类型没有映射，并且此字段未编制索引或不可搜索。如果对 minKey 类型使用相同的字段名称，然后更改为其他类型（例如字符串），则会为第一个非 MinKey 值 OpenSearch 创建动态映射。后续值仍然可以是 Amazon DocumentDB minKey 值。	Amazon DocumentDB 支持 minKey 类型字段。
MaxKey	OpenSearch 可以收录亚马逊 DocumentDB MaxKey 类型的文档。它将该值作为空值保存在文档中。此类型没有映射，并且此字段未编制索引或不可搜索。如果对 MaxKey 类型使用相同的字段名称，然后更改为其他类型（例如字符串），则会为第一个非 MaxKey 值 OpenSearch 创建动态映射。后续值仍然可以是 Amazon DocumentDB maxKey 值。	Amazon DocumentDB 支持 maxKey 类型字段。

我们建议您在摄取管道中配置死信队列 (DLQ)。 OpenSearch 如果您已配置队列，S OpenSearch ervice 会将所有因动态映射失败而无法载入的失败文档发送到队列。

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

限制

在为 Amazon DocumentDB 设置 OpenSearch 摄取管道时，请考虑以下限制：

OpenSearch Ingestion 与 Amazon DocumentDB 的集成目前不支持跨区域提取。您的 Amazon DocumentDB 集群和 OpenSearch 摄取管道必须处于相同的位置。 AWS 区域
OpenSearch Ingestion 与 Amazon DocumentDB 的集成目前不支持跨账户提取。您的 Amazon DocumentDB 集群和 OpenSearch 摄取管道必须处于相同的位置。 AWS 账户
一个 OpenSearch 摄取管道仅支持一个 Amazon DocumentDB 集群作为其来源。
OpenSearch Ingestion 与亚马逊 DocumentDB 的集成特别支持基于亚马逊文档数据库实例的集群。但不支持 Amazon DocumentDB 弹性集群。
OpenSearch Ingestion 集成仅支持 AWS Secrets Manager 作为您的 Amazon DocumentDB 集群的身份验证机制。
无法通过更新现有管道配置来从不同的数据库或集合摄取数据。相反，必须创建新管道。

CloudWatch 警报	描述
`<pipeline-name>`.docucmentdb.凭证已更改	该指标表示 AWS 密钥的轮换频率。
`<pipeline-name>`.docmentdb。 executorRefreshErrors	该指标指示刷新 AWS 密钥失败。
`<pipeline-name>`.docmentdb。 exportRecordsTotal	该指标指示从 Amazon DocumentDB 导出的记录数量。
`<pipeline-name>`.docmentdb。 exportRecordsProcessed	该指标表示 OpenSearch 摄取管道处理的记录数。
`<pipeline-name>`.docmentdb。 exportRecordProcessing错误	该指标表示从 Amazon DocumentDB 集群读取数据时 OpenSearch 摄取管道中出现的处理错误数量。
`<pipeline-name>`.docmentdb。 exportRecordsSuccess总计	该指标指示成功处理的导出记录总数。
`<pipeline-name>`.docmentdb。 exportRecordsFailed总计	该指标指示处理失败的导出记录总数。
`<pipeline-name>`.docucmentdb.bytes 已接收	该指标表示 OpenSearch 摄取管道接收的总字节数。
`<pipeline-name>`.docucmentdb.bytes已处理	该指标表示 OpenSearch 摄取管道处理的总字节数。
`<pipeline-name>`.docmentdb。 exportPartitionQuery总计	该指标指示导出分区总数。
`<pipeline-name>`.docmentdb。 streamRecordsSuccess总计	该指标指示该流中成功处理的记录总数。
`<pipeline-name>`.docmentdb。 streamRecordsFailed总计	该指标指示该流中处理失败的记录总数。

选择您的 Cookie 首选项

自定义 Cookie 首选项

关键

性能

功能

广告

无法保存 Cookie 首选项

在 Amazon DocumentDB 中使用 OpenSearch 摄取管道

先决条件

步骤 1：配置管道角色

步骤 2：创建管道

数据一致性

映射数据类型

限制

推荐的 CloudWatch 警报

下一主题：

上一主题：

需要帮助吗？

本页内容

Related resources

此页内容对您是否有帮助？

Related resources