在 Amaz OpenSearch on DocumentDB 中使用采集管道

您可以将 OpenSearch 采集管道与 Amazon DocumentDB 配合使用，将文档更改（例如创建、更新和删除）流式传输到 OpenSearch 亚马逊服务域和馆藏。如果您的Amaz OpenSearch on DocumentDB集群上可用，摄取管道可以利用变更数据捕获 (CDC) 机制，API也可以利用轮询来提供一种高规模、低延迟的方式来持续流式传输来自Amazon DocumentDB集群的数据。

您可以通过两种方式使用 Amazon DocumentDB 作为数据源来处理数据：有或没有完整的初始快照。

完整的初始快照是对整个 Amazon DocumentDB 集合的批量查询。亚马逊 DocumentDB 将此快照上传到亚马逊 S3。从那里，In OpenSearch gestion 管道将其发送到域中的一个索引，或者将其分区到域中的多个索引。为了保持 Amazon DocumentDB 中的数据并保持 OpenSearch 一致，该管道会将 Amazon DocumentDB 集合中的所有创建、更新和删除事件与保存在一个或多个索引中的文档同步。 OpenSearch

当您使用完整的初始快照时，您的 OpenSearch 摄取管道会首先提取快照，然后开始从 Amazon DocumentDB 变更流中读取数据。它最终会赶上，并保持了 Amazon Doc OpenSearch umentDB 和. 之间近乎实时的数据一致性。

您也可以使用 OpenSearch Ingestion 与 Amazon DocumentDB 的集成在没有快照的情况下直播事件。如果您已经拥有来自其他机制的完整快照，或者您只想使用更改流流流流式传输来自 Amazon DocumentDB 集合的时事，请选择此选项。

使用这两个选项，如果您在工作流配置中启用数据流，则必须在 Amazon DocumentDB 馆藏上启用更改流。如果您只使用满载或导出，则无需启用更改流。

先决条件

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

步骤 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 蓝图来创建此管道。有关更多信息，请参阅 使用蓝图创建管道。

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

要提供自定义CIDR，请从下拉菜单中选择 “其他”。为避免 OpenSearch Ingestion 和 Amazon DocumentDB 之间的 IP 地址冲突，请确保亚马逊 DocumentDB 与 for Ingestion 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 如何自动映射各种数据类型。

{
 "template": {
  "mappings": {
   "properties": {
    "MixedNumberField": {
     "type": "float"
    }
   }
  }
 }
}

{
 "template": {
  "mappings": {
   "properties": {
    "MixedNumberField": {
     "type": "float"
    }
   }
  }
 }
}

{
 "template": {
  "mappings": {
   "properties": {
    "PartType": {
     "type": "keyword"
    }
   }
  }
 }
}

{
 "template": {
  "mappings": {
   "properties": {
    "MixedNumberField": {
     "type": "float"
    }
   }
  }
 }
}

{
 "template": {
  "mappings": {
   "properties": {
    "myDateField": {
     "type": "date",
     "format": "epoch_second"
    }
   }
  }
 }
}

{
 "template": {
  "mappings": {
   "properties": {
    "myTimestampField": {
     "type": "date",
     "format": "epoch_second"
    }
   }
  }
 }
}

{
 "template": {
  "mappings": {
   "properties": {
    "myDecimalField": {
     "type": "double"
    }
   }
  }
 }
}

{
 "template": {
  "mappings": {
   "properties": {
    "imageData": {
     "type": "binary"
    }
   }
  }
 }
}

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

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

限制

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