本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。 # 将 Amazon OpenSearch Ingestion 管道与其他服务和应用程序集成 要成功将数据提取到 Amazon OpenSearch Ingestion 管道，您必须将您的客户端应用程序（*源*）配置为将数据发送到管道终端节点。您的来源可能是客户端，例如 Fluent Bit 日志、 OpenTelemetry 收集器或简单的 S3 存储桶。每个客户端的确切配置各不相同。 源配置期间（与直接向 OpenSearch 服务域或 OpenSearch 无服务器集合发送数据相比）的重要区别在于 AWS 服务名称 (`osis`) 和主机端点，它们必须是管道端点。 ## 构建摄取端点 要将数据摄入管道，请将其发送到摄取端点。要查找摄取 URL，请导航到**管道设置**页面并复制**摄取 URL**。 ![\[Pipeline settings page showing details like status, capacity, and ingestion URL for data input.\]](http://docs.aws.amazon.com/zh_cn/opensearch-service/latest/developerguide/images/pipeline-endpoint.png) 要为基于拉取的来源（例如[OTel 跟踪](https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sources/otel-trace/)和[OTel 指标](https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sources/otel-metrics-source/)）构建完整的摄取终端节点，请将管道配置中的提取路径添加到摄取网址。 例如，假设管道配置的摄取路径如下所示： ![\[Input field for HTTP source path with example "/my/test_path" entered.\]](http://docs.aws.amazon.com/zh_cn/opensearch-service/latest/developerguide/images/ingestion-path.png) 您在客户端配置中指定的完整摄取端点将采用以下格式：`https://ingestion-pipeline-abcdefg.us-east-1.osis.amazonaws.com/my/test_path`。 ## 创建摄取角色 所有对 OpenSearch Ingestion 的请求都必须使用签[名版本 4 进行签名](https://docs.aws.amazon.com/general/latest/gr/signature-version-4.html)。至少，必须向签署请求的角色授予`osis:Ingest`操作权限，从而允许其向 OpenSearch 摄取管道发送数据。 例如，以下 AWS Identity and Access Management (IAM) 策略允许相应的角色向单个管道发送数据： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": "osis:Ingest", "Resource": "arn:aws:osis:us-east-1:111122223333:pipeline/pipeline-name" } ] } ``` ------ **注意** 要将角色用于*所有*管道，请将 `Resource` 元素的 ARN 替换为通配符 (\$1)。 ### 提供跨账户摄取访问权限 **注意** 只能为公有管道（而非 VPC 管道）提供跨账户摄取访问权限。 您可能需要将来自其他渠道的数据提取到管道中 AWS 账户，例如存放源应用程序的帐户。如果写入管道的主体与管道本身的账户不同，则需要将主体配置为信任另一个 IAM 角色，以将数据摄取到管道中。 **配置跨账户摄取权限** 1. 在与管道相同的范围内创建具有`osis:Ingest`权限的摄取角色（ AWS 账户 如上一节所述）。有关说明，请参阅[创建 IAM 角色](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create.html)。 1. 为摄取角色附加[信任策略](https://docs.aws.amazon.com/IAM/latest/UserGuide/roles-managingrole-editing-console.html#roles-managingrole_edit-trust-policy)，允许其他账户主体担任此角色： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [{ "Effect": "Allow", "Principal": { "AWS": "arn:aws:iam::111122223333:root" }, "Action": "sts:AssumeRole" }] } ``` ------ 1. 在另一个账户中，配置您的客户端应用程序（例如，Fluent Bit）担任摄取角色。为使配置生效，应用程序账户必须向应用程序用户或角色授予担任摄取角色的权限。 以下基于身份的示例策略允许附加主体担任管道账户的 `ingestion-role`： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": "sts:AssumeRole", "Resource": "arn:aws:iam::111122223333:role/ingestion-role" } ] } ``` ------ 然后，客户端应用程序可以使用该[AssumeRole](https://docs.aws.amazon.com/STS/latest/APIReference/API_AssumeRole.html)操作来假设数据`ingestion-role`并将其摄取到关联的管道中。 # 在 Atlas OpenSearch sian 服务中使用采集管道 你可以使用 Atlassian Jira 和 Confluence 源插件将来自 Atlassian 服务的数据提取到你的摄取管道中。 OpenSearch 这些集成功能可通过同步完整的 Jira 项目和 Confluence 空间，帮助您创建统一的可搜索知识库，同时借助持续监控和自动更新同步功能保持实时相关性。 ------ #### [ Integrating with Jira ] 将您的 Jira 内容集成到，借助强大的上下文搜索功能，改变您的 Jira 体验。 OpenSearchData Prepper [Atlassian Jira](https://www.atlassian.com/software/jira) 源代码插件可通过同步完整的 Jira 项目，帮助您创建统一的可搜索知识库，同时借助持续监控和自动更新同步功能保持实时相关性。此集成支持数据同步功能，并提供灵活的筛选选项，可针对特定项目、问题类型及状态进行筛选，确保仅导入您所需的信息。 为了确保安全可靠的连接，该插件支持多种身份验证方法，包括基本的 API 密钥身份验证和 OAuth2身份验证，并通过使用存储在中的密钥管理凭据增加了安全性 AWS Secrets Manager。该插件还提供自动令牌续订功能，确保访问不中断，保障持续运行。此集成基于 Atlassian 的 [API 版本 2](https://developer.atlassian.com/cloud/jira/platform/rest/v2/intro/#version%22%3Eapi-version-2)，使团队能够通过高级搜索功能从 Jira 数据中解锁宝贵 OpenSearch的见解。 ------ #### [ Integrating with Confluence ] OpenSearch 通过 Data Prepper 的 Confluence 源插件将 [Atlassian Confluence 内容集成到 Confluenc](https://www.atlassian.com/software/confluence) e 源插件中，增强团队的知识管理和协作能力。此集成使您能够创建集中化、可搜索的集体知识存储库，从而提升信息检索效率并提高团队生产力。通过同步 Confluence 内容并持续监控更新，该插件可确保您的 OpenSearch 索引保持 up-to-date完整且完整。 该集成提供灵活的筛选选项，支持您有选择地导入特定空间或页面类型的内容，从而根据组织需求定制同步内容。该插件支持基本的 API 密钥和 OAuth2 身份验证方法，可以选择通过安全地管理凭据 AWS Secrets Manager。该插件的自动令牌续订功能可确保无中断访问和无缝运行。此集成基于 Atlassian 的 Confluence [API](https://developer.atlassian.com/cloud/confluence/rest/v1/intro/#auth)，使团队能够在其 Confluence 内容中利用 OpenSearch高级搜索功能，从而提高组织内部的信息可访问性和利用率。 ------ **Topics** + [先决条件](#atlassian-prerequisites) + [配置管道角色](#atlassian-pipeline-role) + [Jira 连接器管道配置](#jira-connector-pipeline) + [Confluence 连接器管道配置](#confluence-connector-pipeline) + [数据一致性](#data-consistency) + [限制](#limitations) + [Atlass CloudWatch ian 连接器的指标](#metrics) + [使用 2.0 将 Amazon OpenSearch Ingestion 管道连接到 Atlassian Jira 或 Confluence OAuth](configure-client-atlassian-OAuth2-setup.md) ## 先决条件 在创建 OpenSearch 摄取管道之前，请完成以下步骤： 1. 选择以下选项之一，为您的 Jira 网站准备证书。 OpenSearch 摄取只需要对内容进行`ReadOnly`授权。 1. **选项 1：API 密钥**：登录您的 Atlassian 账户，并使用以下主题中的信息生成 API 密钥： + [管理 Atlassian 账户的 API 令牌](https://support.atlassian.com/atlassian-account/docs/manage-api-tokens-for-your-atlassian-account/) 1. **选项 2： OAuth2**— 登录您的 Atlassian 帐户并使用中的信息。[使用 2.0 将 Amazon OpenSearch Ingestion 管道连接到 Atlassian Jira 或 Confluence OAuth](configure-client-atlassian-OAuth2-setup.md) 1. [在 AWS Secrets Manager中创建密钥](https://docs.aws.amazon.com/secretsmanager/latest/userguide/create_secret.html)，以存储在上一步中创建的凭证。按照步骤进行操作时，作出以下选择： + 对于**密钥类型**，请选择**其他密钥类型**。 + 对于**键/值对**，根据您选择的授权类型创建以下键值对： ------ #### [ API key ] ``` { "username": user-name-usualy-email-id, "password": api-key } ``` ------ #### [ OAuth 2.0 ] ``` { "clientId": client-id "clientSecret": client-secret "accessKey": access-key "refreshKey": refresh-key } ``` ------ 创建密钥后，复制该密钥的 Amazon 资源名称（ARN）。您需要将该密钥包含在管道角色权限策略中。 ## 配置管道角色 传递到管道的角色必须附加以下策略，才能读取和写入先决条件部分创建的密钥。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "SecretReadWrite", "Effect": "Allow", "Action": [ "secretsmanager:GetResourcePolicy", "secretsmanager:GetSecretValue", "secretsmanager:DescribeSecret", "secretsmanager:PutSecretValue", "secretsmanager:ListSecretVersionIds" ], "Resource": "arn:aws:secretsmanager:us-east-1:111122223333:secret:secret-name-random-6-characters" } ] } ``` ------ 该角色还应附带一项策略，用于访问和写入您选择的接收器。例如，如果您选择 OpenSearch 作为接收器，则策略将类似于以下内容： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "OpenSearchWritePolicy", "Effect": "Allow", "Action": "aoss:*", "Resource": "arn:aws:aoss:us-east-1:111122223333:collection/collection-id" } ] } ``` ------ ## Jira 连接器管道配置 您可以使用预先配置的 Atlassian Jira 蓝图，以创建此管道。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 将 *placeholder values* 替换为您自己的信息。 ``` version: "2" extension: aws: secrets: jira-account-credentials: secret_id: "secret-arn" region: "secret-region" sts_role_arn: "arn:aws:iam::123456789012:role/Example-Role" atlassian-jira-pipeline: source: jira: # We only support one host url for now hosts: ["jira-host-url"] acknowledgments: true authentication: # Provide one of the authentication method to use. Supported methods are 'basic' and 'oauth2'. # For basic authentication, password is the API key that you generate using your jira account basic: username: ${{aws_secrets:jira-account-credentials:username}} password: ${{aws_secrets:jira-account-credentials:password}} # For OAuth2 based authentication, we require the following 4 key values stored in the secret # Follow atlassian instructions at the below link to generate these keys. # https://developer.atlassian.com/cloud/confluence/oauth-2-3lo-apps/ # If you are using OAuth2 authentication, we also require, write permission to your AWS secret to # be able to write the renewed tokens back into the secret. # oauth2: # client_id: ${{aws_secrets:jira-account-credentials:clientId}} # client_secret: ${{aws_secrets:jira-account-credentials:clientSecret}} # access_token: ${{aws_secrets:jira-account-credentials:accessToken}} # refresh_token: ${{aws_secrets:jira-account-credentials:refreshToken}} filter: project: key: include: # This is not project name. # It is an alphanumeric project key that you can find under project details in Jira. - "project-key" - "project-key" # exclude: # - "project-key" # - "project-key" issue_type: include: - "issue-type" # - "Story" # - "Bug" # - "Task" # exclude: # - "Epic" status: include: - "ticket-status" # - "To Do" # - "In Progress" # - "Done" # exclude: # - "Backlog" sink: - opensearch: # Provide an Amazon OpenSearch Service domain endpoint hosts: [ "https://search-mydomain-1a2a3a4a5a6a7a8a9a0a9a8a7a.us-east-1.es.amazonaws.com" ] index: "index_${getMetadata(\"project\")}" # Ensure adding unique document id which is the unique ticket id in this case document_id: '${/id}' aws: # Provide a Role ARN with access to the domain. This role should have a trust relationship with osis-pipelines.amazonaws.com sts_role_arn: "arn:aws:iam::123456789012:role/Example-Role" # Provide the region of the domain. region: "us-east-1" # Enable the 'serverless' flag if the sink is an Amazon OpenSearch Serverless collection serverless: false # serverless_options: # Specify a name here to create or update network policy for the serverless collection # network_policy_name: "network-policy-name" # Enable the 'distribution_version' setting if the Amazon OpenSearch Service domain is of version Elasticsearch 6.x # distribution_version: "es6" # Enable and switch the 'enable_request_compression' flag if the default compression setting is changed in the domain. # See 在亚马逊 OpenSearch 服务中压缩 HTTP 请求 # enable_request_compression: true/false # Optional: Enable the S3 DLQ to capture any failed requests in an S3 bucket. Delete this entire block if you don't want a DLQ. dlq: s3: # Provide an S3 bucket bucket: "your-dlq-bucket-name" # Provide a key path prefix for the failed requests # key_path_prefix: "kinesis-pipeline/logs/dlq" # Provide the region of the bucket. region: "us-east-1" # Provide a Role ARN with access to the bucket. This role should have a trust relationship with osis-pipelines.amazonaws.com sts_role_arn: "arn:aws:iam::123456789012:role/Example-Role" ``` Jira 源中属性的键： 1. **主机**：Jira 云或本地 URL。通常类似 `https://your-domain-name.atlassian.net/`。 1. **确认**：确保数据能够完整无误地传输至接收端。 1. **身份验证**：描述您希望管道访问 Jira 实例的方式。选择`Basic`或`OAuth2`并指定引用密钥中密钥的相应密钥属性。 AWS 1. **筛选器**：此部分可帮助您选择要提取和同步的 Jira 数据范围。 1. **项目**：在 `include` 部分列出要同步的项目密钥。否则，在 `exclude` 部分下方列出要排除的项目。每次仅提供包含或排除选项中的一种。 1. **issue\$1type**：需要同步的特定问题类型。根据需要类似的 `include` 或 `exclude` 模式。请注意，附件将显示为指向原始附件的锚点链接，但不会提取附件内容。 1. **状态**：要应用于数据提取查询的特定状态筛选器。如果指定 `include`，则仅同步具有这些状态的票证。如果指定 `exclude`，则所有票证（除列出的处于排除状态的票证外）都将同步。 ## Confluence 连接器管道配置 您可以使用预先配置的 Atlassian Confluence 蓝图，以创建此管道。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 ``` version: "2" extension: aws: secrets: confluence-account-credentials: secret_id: "secret-arn" region: "secret-region" sts_role_arn: "arn:aws:iam::123456789012:role/Example-Role" atlassian-confluence-pipeline: source: confluence: # We currently support only one host URL. hosts: ["confluence-host-url"] acknowledgments: true authentication: # Provide one of the authentication method to use. Supported methods are 'basic' and 'oauth2'. # For basic authentication, password is the API key that you generate using your Confluence account basic: username: ${{aws_secrets:confluence-account-credentials:confluenceId}} password: ${{aws_secrets:confluence-account-credentials:confluenceCredential}} # For OAuth2 based authentication, we require the following 4 key values stored in the secret # Follow atlassian instructions at the following link to generate these keys: # https://developer.atlassian.com/cloud/confluence/oauth-2-3lo-apps/ # If you are using OAuth2 authentication, we also require write permission to your AWS secret to # be able to write the renewed tokens back into the secret. # oauth2: # client_id: ${{aws_secrets:confluence-account-credentials:clientId}} # client_secret: ${{aws_secrets:confluence-account-credentials:clientSecret}} # access_token: ${{aws_secrets:confluence-account-credentials:accessToken}} # refresh_token: ${{aws_secrets:confluence-account-credentials:refreshToken}} filter: space: key: include: # This is not space name. # It is a space key that you can find under space details in Confluence. - "space key" - "space key" # exclude: # - "space key" # - "space key" page_type: include: - "content type" # - "page" # - "blogpost" # - "comment" # exclude: # - "attachment" sink: - opensearch: # Provide an Amazon OpenSearch Service domain endpoint hosts: [ "https://search-mydomain-1a2a3a4a5a6a7a8a9a0a9a8a7a.us-east-1.es.amazonaws.com" ] index: "index_${getMetadata(\"space\")}" # Ensure adding unique document id which is the unique ticket ID in this case. document_id: '${/id}' aws: # Provide the Amazon Resource Name (ARN) for a role with access to the domain. This role should have a trust relationship with osis-pipelines.amazonaws.com. sts_role_arn: "arn:aws:iam::123456789012:role/Example-Role" # Provide the Region of the domain. region: "us-east-1" # Enable the 'serverless' flag if the sink is an Amazon OpenSearch Serverless collection serverless: false # serverless_options: # Specify a name here to create or update network policy for the serverless collection. # network_policy_name: "network-policy-name" # Enable the 'distribution_version' setting if the Amazon OpenSearch Service domain is of version Elasticsearch 6.x # distribution_version: "es6" # Enable and switch the 'enable_request_compression' flag if the default compression setting is changed in the domain. # For more information, see 在亚马逊 OpenSearch 服务中压缩 HTTP 请求. # enable_request_compression: true/false # Optional: Enable the S3 DLQ to capture any failed requests in an S3 bucket. Delete this entire block if you don't want a DLQ. dlq: s3: # Provide an S3 bucket bucket: "your-dlq-bucket-name" # Provide a key path prefix for the failed requests # key_path_prefix: "kinesis-pipeline/logs/dlq" # Provide the Rregion of the bucket. region: "us-east-1" # Provide the Amazon Resource Name (ARN) for a role with access to the bucket. This role should have a trust relationship with osis-pipelines.amazonaws.com sts_role_arn: "arn:aws:iam::123456789012:role/Example-Role" ``` Confluence 源中的键属性： 1. **主机**：Confluence 云或本地 URL。通常类似 `https://your-domain-name.atlassian.net/` 1. **确认**：确保数据能够完整无误地传输至接收端。 1. **身份验证**：描述您希望管道访问 Confluence 实例的方式。选择`Basic`或`OAuth2`并指定引用密钥中密钥的相应密钥属性。 AWS 1. **筛选器**：此部分可帮助您选择要提取和同步的 Confluence 数据范围。 1. **空间**：在 `include` 部分列出要同步的空间密钥。否则，在 `exclude` 部分下方列出要排除的空间。每次仅提供包含或排除选项中的一种。 1. **page\$1type**：希望同步的特定页面类型（如页面、博客文章或附件）。根据需要类似的 `include` 或 `exclude` 模式。请注意，附件将显示为指向原始附件的锚点链接，但不会提取附件内容。 ## 数据一致性 根据管道 YAML 中指定的筛选器，选定的项目（或空间）将被提取一次，并完全同步到目标接收器。然后，持续变更监控将实时捕获变更并更新接收端的数据。一个例外情况是，变更监控仅同步 `create` 和 `update` 操作，而不会同步 `delete` 操作。 ## 限制 + 用户删除操作不会同步。数据一旦记录在接收器中，就会保留在接收器中。如果在接收器设置中指定了 ID 映射，更新操作将使用新更改覆盖现有内容。 + 使用不支持以下内容的旧版 Atlassian 软件的本地实例与该来源 APIs 不兼容： + Jira Search API 版本 3 + `rest/api/3/search` + `rest/api/3/issue` + Confluence + `wiki/rest/api/content/search` + `wiki/rest/api/content` + `wiki/rest/api/settings/systemInfo` ## Atlass CloudWatch ian 连接器的指标 **类型：Jira 连接器指标** | 来源 | 指标 | 指标类型 | | --- | --- | --- | | acknowledgementSetSuccesses。count | 计数器 | 如果已启用确认功能，此指标将提供同步成功的票证数量。 | | acknowledgementSetFailures。count | 计数器 | 如果已启用确认功能，此指标将提供同步失败的票证数量。 | | crawlingTime.avg | 计时器 | 浏览所有新增内容所需的时间。 | | ticketFetchLatency.avg | 计时器 | 票证获取 API 延迟平均值。 | | ticketFetchLatency.max | 计时器 | 票证获取 API 延迟最大值。 | | ticketsRequested.count | 计数器 | 已发送的票证获取请求数量。 | | ticketRequestedFailed。count | 计数器 | 未发送的票证获取请求数量。 | | ticketRequestedSuccess。count | 计数器 | 成功的票证获取请求数量。 | | searchCallLatency.avg | 计时器 | 搜索 API 调用延迟平均值。 | | searchCallLatency.max | 计时器 | 搜索 API 调用延迟最大值。 | | searchResultsFound。count | 计数器 | 给定搜索调用中找到的项目数。 | | searchRequestFailed。count | 计数器 | 搜索 API 调用失败计数。 | | authFailures.count | 计数器 | 身份验证失败计数。 | **类型：Confluence 连接器指标** | 来源 | 指标 | 指标类型 | | --- | --- | --- | | acknowledgementSetSuccesses。count | 计数器 | 如果已启用确认功能，此指标将提供同步成功的页面数量。 | | acknowledgementSetFailures。count | 计数器 | 如果已启用确认功能，此指标将提供同步失败的页面数量。 | | crawlingTime.avg | 计时器 | 浏览所有新增内容所需的时间。 | | pageFetchLatency.avg | 计时器 | 内容获取 API 延迟（平均值）。 | | pageFetchLatency.max | 计时器 | 内容获取 API 延迟（最大值）。 | | pagesRequested.count | 计数器 | 内容获取 API 的调用次数。 | | pageRequestFailed。count | 计数器 | 内容获取 API 的失败请求数。 | | pageRequestedSuccess。count | 计数器 | 内容获取 API 的成功请求数。 | | searchCallLatency.avg | 计时器 | 搜索 API 调用延迟平均值。 | | searchCallLatency.max | 计时器 | 搜索 API 调用延迟最大值 | | searchResultsFound。count | 计数器 | 给定搜索调用中找到的项目数。 | | searchRequestsFailed。count | 计数器 | 搜索 API 调用失败计数。 | | authFailures.count | 计数器 | 身份验证失败计数。 | # 使用 2.0 将 Amazon OpenSearch Ingestion 管道连接到 Atlassian Jira 或 Confluence OAuth 使用本主题中的信息来帮助您使用 2.0 身份验证配置 Amazon OpenSearch Ingestion 管道并将其连接到 Jira 或 Confluence 账户。 OAuth 在完成使用 Atlassian S [先决条件](configure-client-atlassian.md#atlassian-prerequisites) ervices 的 OpenSearch 摄取管道但选择不使用 API 密钥凭证时执行此任务。 **Topics** + [创建 OAuth 2.0 集成应用程序](#create-OAuth2-integration-app) + [生成和刷新 Atlassian Developer 访问令牌](#generate-and-refresh-jira-access-token) ## 创建 OAuth 2.0 集成应用程序 使用以下步骤帮助您在 Atlassian 开发者网站上创建 OAuth 2.0 集成应用程序。 **创建 OAuth 2.0 集成应用程序** 1. [通过 myapps/ 登录您的 Atlassian 开发者账户。 https://developer.atlassian.com/console/](https://developer.atlassian.com/console/myapps/) 1. 选择**创建**，**OAuth 2.0 集成**。 1. 对于**名称**，输入名称以标识应用程序的用途。 1. 选择**我同意遵守 Atlassian 的开发人员条款**复选框，然后选择**创建**。 1. 在左侧导航栏中，选择**授权**，然后选择**添加**。 1. 在**回调 URL** 中，输入任意 URL，例如 **https://www.amazon.com** 或 **https://www.example.com**，然后选择**保存更改**。 1. 在左侧导航栏中，选择**权限**页面，然后在 Jira API 的行中选择**添加**，接着选择**配置**。勾选所有经典范围的读取权限（列表如下），然后选择“保存” 1. 选择**精细范围**选项卡，然后选择**编辑范围**，以打开**编辑 Jira API** 对话框。 1. 选择正在使用的源插件的权限： ------ #### [ Jira ] ``` read:audit-log:jira read:issue:jira read:issue-meta:jira read:attachment:jira read:comment:jira read:comment.property:jira read:field:jira read:field.default-value:jira read:field.option:jira read:field-configuration-scheme:jira read:field-configuration:jira read:issue-link:jira read:issue-link-type:jira read:issue-link-type:jira read:issue.remote-link:jira read:issue.property:jira read:resolution:jira read:issue-details:jira read:issue-type:jira read:issue-worklog:jira read:issue-field-values:jira read:issue.changelog:jira read:issue.transition:jira read:issue.vote:jira read:jira-expressions:jira ``` ------ #### [ Confluence ] ``` read:content:confluence read:content-details:confluence read:space-details:confluence read:audit-log:confluence read:page:confluence read:blogpost:confluence read:custom-content:confluence read:comment:confluence read:space:confluence read:space.property:confluence read:space.setting:confluence read:content.property:confluence read:content.metadata:confluence read:task:confluence read:whiteboard:confluence read:app-data:confluence manage:confluence-configuration ``` ------ 1. 选择**保存**。 有关相关信息，请参阅 At [lassian 开发者网站上的实施 OAuth 2.0 (3LO)](https://developer.atlassian.com/cloud/oauth/getting-started/implementing-oauth-3lo/) [和确定操作所需的范围](https://developer.atlassian.com/cloud/oauth/getting-started/determining-scopes/)。 ## 生成和刷新 Atlassian Developer 访问令牌 使用以下步骤，以帮助您在 Atlassian Developer 网站上生成和刷新 Atlassian Developer 访问令牌。 **生成和刷新 Jira 访问令牌** 1. [通过 myapps/ 登录您的 Atlassian 开发者账户。 https://developer.atlassian.com/console/](https://developer.atlassian.com/console/myapps/) 1. 选择在 [创建 OAuth 2.0 集成应用程序](#create-OAuth2-integration-app) 中创建的应用程序。 1. 从左侧导航中，选择**授权**。 1. 从页面底部复制精细的 Atlassian API 授权 URL 值，然后将其粘贴到您选择的文本编辑器中。 URL 的格式如下所示： ``` https://auth.atlassian.com/authorize? audience=api.atlassian.com &client_id=YOUR_CLIENT_ID &scope=REQUESTED_SCOPE%20REQUESTED_SCOPE_TWO &redirect_uri=https://YOUR_APP_CALLBACK_URL &state=YOUR_USER_BOUND_VALUE &response_type=code &prompt=consent ``` 1. 对于 `state=YOUR_USER_BOUND_VALUE`，将参数值更改为您选择的任何值，例如 state="**sample\$1text**"。 有关更多信息，请参阅 Atlassian Developer 网站上的 [What is the state parameter used for?](https://developer.atlassian.com/cloud/jira/platform/oauth-2-3lo-apps/#what-is-the-state-parameter-used-for-)。 1. 请注意，`scope` 部分列出您在先前任务中选择的精细范围。例如：`scope=read%3Ajira-work%20read%3Ajira-user%20offline_access` `offline_access` 表示您要生成 `refresh_token`。 1. 打开 Web 浏览器窗口，在浏览器窗口的地址栏中输入您复制的授权 URL。 1. 目标页面打开后，请确认信息准确无误，然后选择**接受**，以重定向至 Jira 或 Confluence 主页。 1. 主页加载完成后，复制此页面的 URL。该 URL 包含应用程序的授权码。您可使用此代码生成访问令牌。`code=` 之后的整个部分即为授权码。 1. 使用以下 cURL 命令生成访问令牌。将 *placeholder values* 替换为您自己的信息。 **提示** 您也可以使用第三方服务，例如 Postman。 ``` curl --request POST --url 'https://auth.atlassian.com/oauth/token' \ --header 'Content-Type: application/json' \ --data '{"grant_type": "authorization_code", "client_id": "YOUR_CLIENT_ID", "client_secret": "YOUR_CLIENT_SECRET", "code": "AUTHORIZATION_CODE", "redirect_uri": "YOUR_CALLBACK_URL"}' ``` 对此命令的响应包括 `access_code` 和 `refresh_token` 的值。 # 在 Amaz OpenSearch on Aurora 上使用采集管道 您可以将 OpenSearch 采集管道与 Amazon Aurora 配合使用，将现有数据和流更改（例如创建、更新和删除）导出到亚马逊 OpenSearch 服务域和集合。 OpenSearch 采集管道整合了变更数据捕获 (CDC) 基础设施，为持续流式传输来自 Amazon Aurora 的数据提供了一种高规模、低延迟的方式。支持 Aurora MySQL 和 Aurora PostgreSQL。 您可以通过两种方式将 Amazon Aurora 作为处理数据的来源：有或没有完整初始快照。完整初始快照是指定表的快照，此快照将导出至 Amazon S3。从那里，In OpenSearch gestion 管道将其发送到域中的一个索引，或者将其分区到域中的多个索引。为了保持 Amazon Aurora 中数据 OpenSearch 的一致性，该管道会将 Amazon Aurora 集群中表中的所有创建、更新和删除事件与保存在一个或多个索引中的 OpenSearch 文档同步。 当您使用完整的初始快照时，您的 OpenSearch 摄取管道会首先提取快照，然后开始从 Amazon Aurora 变更流中读取数据。它最终赶上并保持了 Amazon Aurora 和 Amazon Aurora 之间近乎实时的数据一致性。 OpenSearch 您还可以使用 OpenSearch Ingestion 与 Amazon Aurora 的集成来跟踪变更数据的捕获并将所有更新收录到 Aurora 中。 OpenSearch如果您已通过其他机制获取完整快照，或仅需捕获 Amazon Aurora 集群中数据的所有更改，则选择此选项。 选择此选项时，您需要[为 Aurora MySQL 配置二进制日志记录](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/USER_LogAccess.MySQL.BinaryFormat.html)，或者[在集群上为 Aurora PostgreSQL 设置逻辑复制](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/AuroraPostgreSQL.Replication.Logical.Configure.html)。 **Topics** + [Aurora MySQL](aurora-mysql.md) + [Aurora PostgreSQL](aurora-PostgreSQL.md) # Aurora MySQL 完成以下步骤，使用适用于 Aurora MySQL 的 Amazon Aurora 配置 OpenSearch 摄取管道。 **Topics** + [Aurora MySQL 先决条件](#aurora-mysql-prereqs) + [步骤 1：配置管道角色](#aurora-mysql-pipeline-role) + [步骤 2：创建管道](#aurora-mysql-pipeline) + [数据一致性](#aurora-mysql-pipeline-consistency) + [映射数据类型](#aurora-mysql-pipeline-mapping) + [限制](#aurora-mysql-pipeline-limitations) + [推荐 CloudWatch 警报](#aurora-mysql-pipeline-metrics) ## Aurora MySQL 先决条件 在创建 OpenSearch Ingestion 管道之前，请执行以下步骤： 1. [在 Amazon Aurora 中创建自定义 Aurora 数据库集群参数组以配置二进制日志记录](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/zero-etl.setting-up.html#zero-etl.parameters)。 ``` aurora_enhanced_binlog=1 binlog_backup=0 binlog_format=ROW binlog_replication_globaldb=0 binlog_row_image=full binlog_row_metadata=full ``` 此外，请确保 `binlog_transaction_compression` 参数未设置为 `ON`，`binlog_row_value_options` 参数也未设置为 `PARTIAL_JSON`。 1. [选择或创建 Aurora MySQL 数据库集群](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/CHAP_GettingStartedAurora.CreatingConnecting.Aurora.html)，并将上一步创建的参数组与该数据库集群相关联。 1. [将二进制日志保留时间配置为 24 小时或更长](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/mysql-stored-proc-configuring.html)。 1. 使用 [Aurora 和 AWS Secrets Manager的密码管理](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/rds-secrets-manager.html)，在 Amazon Aurora 集群上设置用户名和密码身份验证。您也可以通过创建 Secrets [Manager 密钥来创建 username/password ](https://docs.aws.amazon.com/secretsmanager/latest/userguide/create_secret.html)组合。 1. 如果您使用完整的初始快照功能，请创建一个 AWS KMS key 和一个 IAM 角色来将数据从 Amazon Aurora 导出到 Amazon S3。 IAM 角色还必须具有以下权限策略： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "ExportPolicy", "Effect": "Allow", "Action": [ "s3:PutObject*", "s3:ListBucket", "s3:GetObject*", "s3:DeleteObject*", "s3:GetBucketLocation" ], "Resource": [ "arn:aws:s3:::s3-bucket-used-in-pipeline", "arn:aws:s3:::s3-bucket-used-in-pipeline/*" ] } ] } ``` ------ 该角色还应拥有以下信任关系： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "export.rds.amazonaws.com" }, "Action": "sts:AssumeRole" } ] } ``` ------ 1. 选择或创建 OpenSearch 服务域或 OpenSearch 无服务器集合。有关更多信息，请参阅[创建 OpenSearch 服务域](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/createupdatedomains.html#createdomains)和[创建集合](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/serverless-manage.html#serverless-create)。 1. 将[基于资源的策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/ac.html#ac-types-resource)附加到域，或将[数据访问策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/serverless-data-access.html)附加到集合。这些访问策略允许 OpenSearch Ingestion 将数据从您的 Amazon Aurora 数据库集群写入您的域或集合。 ## 步骤 1：配置管道角色 完成 Amazon Aurora 管道先决条件设置后，[配置管道角色](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/pipeline-security-overview.html#pipeline-security-sink)，以便在管道配置中使用。还需为该角色添加以下针对 Amazon Aurora 源的权限： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "allowReadingFromS3Buckets", "Effect": "Allow", "Action": [ "s3:GetObject", "s3:DeleteObject", "s3:GetBucketLocation", "s3:ListBucket", "s3:PutObject" ], "Resource": [ "arn:aws:s3:::s3_bucket", "arn:aws:s3:::s3_bucket/*" ] }, { "Sid": "allowNetworkInterfacesActions", "Effect": "Allow", "Action": [ "ec2:AttachNetworkInterface", "ec2:CreateNetworkInterface", "ec2:CreateNetworkInterfacePermission", "ec2:DeleteNetworkInterface", "ec2:DeleteNetworkInterfacePermission", "ec2:DetachNetworkInterface", "ec2:DescribeNetworkInterfaces" ], "Resource": [ "arn:aws:ec2:*:111122223333:network-interface/*", "arn:aws:ec2:*:111122223333:subnet/*", "arn:aws:ec2:*:111122223333:security-group/*" ] }, { "Sid": "allowDescribeEC2", "Effect": "Allow", "Action": [ "ec2:Describe*" ], "Resource": "*" }, { "Sid": "allowTagCreation", "Effect": "Allow", "Action": [ "ec2:CreateTags" ], "Resource": "arn:aws:ec2:*:111122223333:network-interface/*", "Condition": { "StringEquals": { "aws:RequestTag/OSISManaged": "true" } } }, { "Sid": "AllowDescribeInstances", "Effect": "Allow", "Action": [ "rds:DescribeDBInstances" ], "Resource": [ "arn:aws:rds:us-east-2:111122223333:db:*" ] }, { "Sid": "AllowDescribeClusters", "Effect": "Allow", "Action": [ "rds:DescribeDBClusters" ], "Resource": [ "arn:aws:rds:us-east-2:111122223333:cluster:DB-id" ] }, { "Sid": "AllowSnapshots", "Effect": "Allow", "Action": [ "rds:DescribeDBClusterSnapshots", "rds:CreateDBClusterSnapshot", "rds:AddTagsToResource" ], "Resource": [ "arn:aws:rds:us-east-2:111122223333:cluster:DB-id", "arn:aws:rds:us-east-2:111122223333:cluster-snapshot:DB-id*" ] }, { "Sid": "AllowExport", "Effect": "Allow", "Action": [ "rds:StartExportTask" ], "Resource": [ "arn:aws:rds:us-east-2:111122223333:cluster:DB-id", "arn:aws:rds:us-east-2:111122223333:cluster-snapshot:DB-id*" ] }, { "Sid": "AllowDescribeExports", "Effect": "Allow", "Action": [ "rds:DescribeExportTasks" ], "Resource": "*", "Condition": { "StringEquals": { "aws:RequestedRegion": "us-east-2", "aws:ResourceAccount": "111122223333" } } }, { "Sid": "AllowAccessToKmsForExport", "Effect": "Allow", "Action": [ "kms:Decrypt", "kms:Encrypt", "kms:DescribeKey", "kms:RetireGrant", "kms:CreateGrant", "kms:ReEncrypt*", "kms:GenerateDataKey*" ], "Resource": [ "arn:aws:kms:us-east-2:111122223333:key/export-key-id" ] }, { "Sid": "AllowPassingExportRole", "Effect": "Allow", "Action": "iam:PassRole", "Resource": [ "arn:aws:iam::111122223333:role/export-role" ] }, { "Sid": "SecretsManagerReadAccess", "Effect": "Allow", "Action": [ "secretsmanager:GetSecretValue" ], "Resource": [ "arn:aws:secretsmanager:*:111122223333:secret:*" ] } ] } ``` ------ ## 步骤 2：创建管道 配置类似于以下 OpenSearch 内容的摄取管道。示例管道将 Amazon Aurora 集群指定为源。 ``` version: "2" aurora-mysql-pipeline: source: rds: db_identifier: "cluster-id" engine: aurora-mysql database: "database-name" tables: include: - "table1" - "table2" s3_bucket: "bucket-name" s3_region: "bucket-region" s3_prefix: "prefix-name" export: kms_key_id: "kms-key-id" iam_role_arn: "export-role-arn" stream: true aws: sts_role_arn: "arn:aws:iam::account-id:role/pipeline-role" region: "us-east-1" authentication: username: ${{aws_secrets:secret:username}} password: ${{aws_secrets:secret:password}} sink: - opensearch: hosts: ["https://search-mydomain.us-east-1.es.amazonaws.com"] index: "${getMetadata(\"table_name\")}" index_type: custom document_id: "${getMetadata(\"primary_key\")}" action: "${getMetadata(\"opensearch_action\")}" document_version: "${getMetadata(\"document_version\")}" document_version_type: "external" aws: sts_role_arn: "arn:aws:iam::account-id:role/pipeline-role" region: "us-east-1" extension: aws: secrets: secret: secret_id: "rds-secret-id" region: "us-east-1" sts_role_arn: "arn:aws:iam::account-id:role/pipeline-role" refresh_interval: PT1H ``` 您可以使用预先配置的 Amazon Aurora 蓝图，以创建此管道。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 要使用 Amazon Aurora 作为源，您需要为管道配置 VPC 访问权限。您选择的 VPC 应与 Amazon Aurora 源使用的 VPC 相同。然后选择一个或多个子网以及一个或多个 VPC 安全组。请注意，该管道需要访问 Aurora MySQL 数据库的网络权限，因此您还需确认 Aurora 集群已配置 VPC 安全组，允许来自管道 VPC 安全组的入站流量访问数据库端口。有关更多信息，请参阅[使用安全组控制访问权限](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/Overview.RDSSecurityGroups.html)。 如果您使用创建管道，则还必须将管道连接到 VPC，才能使用 Amazon Aurora 作为来源。 AWS 管理控制台 要执行此操作，请找到**网络配置**部分，选择**连接到 VPC** 复选框，然后从提供的任意一个默认选项中选择 CIDR，或者选择自己的 CIDR。您可以按照 [RFC 1918 当前最佳实践](https://datatracker.ietf.org/doc/html/rfc1918)中的定义，使用私有地址空间中的任何 CIDR。 要提供自定义 CIDR，请从下拉菜单中选择**其他**。为避免 OpenSearch Ingestion 和 Amazon Aurora 之间的 IP 地址冲突，请确保 Amazon Aurora VPC CIDR 与用于摄取的 CIDR 不同。 OpenSearch 有关更多信息，请参阅[为管道配置 VPC 访问权限](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/pipeline-security.html#pipeline-vpc-configure)。 ## 数据一致性 该管道通过持续轮询或接收来自 Amazon Aurora 集群的更改以及更新 OpenSearch索引中的相应文档来确保数据一致性。 OpenSearch Ingestion 支持 end-to-end确认，以确保数据的持久性。管道读取快照或流时，它会动态创建分区以进行并行处理。当管道在摄取 OpenSearch 域或集合中的所有记录后收到确认信息时，该管道会将该分区标记为已完成。如果要收录到 OpenSearch 无服务器搜索集合中，可以在管道中生成文档 ID。如果要采集到 OpenSearch Serverless 时间序列集合，请注意管道不会生成文档 ID，因此您必须在工作流接收器配置`document_id: "${getMetadata(\"primary_key\")}"`中省略文档 ID。 In OpenSearch gestion 管道还将传入的事件操作映射到相应的批量索引操作中，以帮助采集文档。这样可以保持数据的一致性，因此 Amazon Aurora 中的每一次数据更改都与中的 OpenSearch相应文档更改保持一致。 ## 映射数据类型 OpenSearch 摄取管道将 MySQL 数据类型映射到适合 OpenSearch 服务域或集合使用的表示形式。如果中未定义映射模板 OpenSearch，则会根据第一个发送的文档使用[动态映射 OpenSearch ](https://opensearch.org/docs/latest/field-types/#dynamic-mapping)自动确定字段类型。您还可以 OpenSearch 通过映射模板明确定义最适合您的字段类型。 下表列出了 MySQL 数据类型和相应的 OpenSearch 字段类型。 OpenSearch 如果未定义显式映射，则 “*默认 OpenSearch 字段类型*” 列将显示相应的字段类型。在这种情况下，使用动态映射 OpenSearch自动确定字段类型。“*推荐的 OpenSearch 字段类型*” 列是建议在映射模板中明确指定的相应字段类型。这些字段类型与 MySQL 中的数据类型更加一致，通常可以启用中提供的更好的搜索功能 OpenSearch。 | MySQL 数据类型 | 默认 OpenSearch 字段类型 | 推荐的 OpenSearch 字段类型 | | --- | --- | --- | | BIGINT | 长整数 | 长整数 | | BIGINT UNSIGNED | 长整数 | 无符号长整数 | | BIT | 长整数 | 字节、短整数、整数或长整数，具体取决于位数 | | DECIMAL | 文本 | 双精度或关键字 | | DOUBLE | 浮点数 | double | | FLOAT | 浮点数 | 浮点数 | | INT | 长整数 | 整数 | | INT UNSIGNED | 长整数 | 长整数 | | MEDIUMINT | 长整数 | 整数 | | MEDIUMINT UNSIGNED | 长整数 | 整数 | | NUMERIC | 文本 | 双精度或关键字 | | SMALLINT | 长整数 | short | | SMALLINT UNSIGNED | 长整数 | 整数 | | TINYINT | 长整数 | 字节 | | TINYINT UNSIGNED | 长整数 | short | | BINARY | 文本 | binary | | BLOB | 文本 | binary | | CHAR | 文本 | 文本 | | ENUM | 文本 | 关键字 | | LONGBLOB | 文本 | binary | | LONGTEXT | 文本 | 文本 | | MEDIUMBLOB | 文本 | binary | | MEDIUMTEXT | 文本 | 文本 | | SET | 文本 | 关键字 | | TEXT | 文本 | 文本 | | TINYBLOB | 文本 | binary | | TINYTEXT | 文本 | 文本 | | VARBINARY | 文本 | binary | | VARCHAR | 文本 | 文本 | | DATE | 长整数（以 epoch 毫秒为单位） | date | | DATETIME | 长整数（以 epoch 毫秒为单位） | date | | TIME | 长整数（以 epoch 毫秒为单位） | date | | TIMESTAMP | 长整数（以 epoch 毫秒为单位） | date | | YEAR | 长整数（以 epoch 毫秒为单位） | date | | GEOMETRY | 文本（采用 WKT 格式） | geo\$1shape | | GEOMETRYCOLLECTION | 文本（采用 WKT 格式） | geo\$1shape | | LINESTRING | 文本（采用 WKT 格式） | geo\$1shape | | MULTILINESTRING | 文本（采用 WKT 格式） | geo\$1shape | | MULTIPOINT | 文本（采用 WKT 格式） | geo\$1shape | | MULTIPOLYGON | 文本（采用 WKT 格式） | geo\$1shape | | POINT | 文本（采用 WKT 格式） | geo\$1point 或 geo\$1shape | | POLYGON | 文本（采用 WKT 格式） | geo\$1shape | | JSON | 文本 | object | 我们建议您在摄取管道中配置死信队列 (DLQ)。 OpenSearch 如果您已配置队列，S OpenSearch ervice 会将所有因动态映射失败而无法载入的失败文档发送到队列。 如果自动映射失败，则可以在管道配置中使用 `template_type` 和 `template_content`，以定义显式映射规则。或者，您可以在启动管道之前直接在搜索域或集合中创建映射模板。 ## 限制 在为 Aurora MySQL 设置 OpenSearch 摄取管道时，请考虑以下限制： + 该集成仅支持每个管道一个 MySQL 数据库。 + 该集成目前不支持跨区域数据摄取；您的 Amazon Aurora 集群和 OpenSearch 域必须处于相同状态。 AWS 区域 + 该集成目前不支持跨账户数据摄取；您的 Amazon Aurora 集群和 OpenSearch 摄取管道必须处于相同状态。 AWS 账户 + 确保 Amazon Aurora 集群已使用 Secrets Manager 启用身份验证，这是唯一受支持的身份验证机制。 + 无法更新现有的管道配置以从不同的数据库中提取不同 and/or 表的数据。要更新管道的数据库和/或表名称，则必须停止该管道，然后使用更新后的配置重新启动，或者创建新管道。 + 通常不支持数据定义语言（DDL）语句。以下情况下，数据一致性将无法得到保证： + 主键已更改 (add/delete/rename)。 + 表已被删除/截断。 + 列名或数据类型已发生更改。 + 如果要同步的 MySQL 表未定义主键，则无法保证数据一致性。您需要在 OpenSearch 接收器配置中正确定义自定义`document_id`选项才能同步 updates/deletes 到 OpenSearch。 + 不支持带有级联删除操作的外键引用，这可能会导致 Aurora MySQL 和之间的数据不一致。 OpenSearch + 支持的版本：Aurora MySQL 版本 3.05.2 及更高版本 ## 推荐 CloudWatch 警报 建议使用以下 CloudWatch 指标来监控您的摄取管道的性能。这些指标可帮助您确定处理的导出数据量、处理的流事件量、处理导出和流事件时的错误数以及写入目标的文档数量。您可以设置 CloudWatch 警报，以便在其中一个指标在指定时间内超过指定值时执行操作。 | 指标 | 说明 | | --- | --- | | pipeline-name.rds.Credentals已更改 | 该指标表示 AWS 密钥的轮换频率。 | | pipeline-name.rds。 executorRefreshErrors | 此指标表示无法刷新 AWS 密钥。 | | pipeline-name.rds。 exportRecordsTotal | 该指标指示从 Amazon Aurora 导出的记录数量。 | | pipeline-name.rds。 exportRecordsProcessed | 该指标表示 OpenSearch 摄取管道处理的记录数。 | | pipeline-name.rds。 exportRecordProcessing错误 | 该指标表示从 Amazon Aurora 集群读取数据时 OpenSearch 摄取管道中出现的处理错误数量。 | | pipeline-name.rds。 exportRecordsSuccess总计 | 该指标指示成功处理的导出记录总数。 | | pipeline-name.rds。 exportRecordsFailed总计 | 该指标指示处理失败的导出记录总数。 | | pipeline-name已收到.rds.bytes | 该指标表示 OpenSearch 摄取管道接收的总字节数。 | | pipeline-name.rds.bytes已处理 | 该指标表示 OpenSearch 摄取管道处理的总字节数。 | | pipeline-name.rds。 streamRecordsSuccess总计 | 该指标指示该流中成功处理的记录总数。 | | pipeline-name.rds。 streamRecordsFailed总计 | 该指标指示该流中处理失败的记录总数。 | # Aurora PostgreSQL 完成以下步骤，使用适用于 Aurora Post OpenSearch greSQL 的 Amazon Aurora 配置摄取管道。 **Topics** + [Aurora PostgreSQL 先决条件](#aurora-PostgreSQL-prereqs) + [步骤 1：配置管道角色](#aurora-mysql-pipeline-role) + [步骤 2：创建管道](#aurora-PostgreSQL-pipeline) + [数据一致性](#aurora-mysql-pipeline-consistency) + [映射数据类型](#aurora-PostgreSQL-pipeline-mapping) + [限制](#aurora-PostgreSQL-pipeline-limitations) + [推荐 CloudWatch 警报](#aurora-mysql-pipeline-metrics) ## Aurora PostgreSQL 先决条件 在创建 OpenSearch Ingestion 管道之前，请执行以下步骤： 1. 在 Amazon Aurora 中[创建自定义数据库集群参数组](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/CHAP_GettingStartedAurora.CreatingConnecting.Aurora.html)以配置逻辑复制。 ``` rds.logical_replication=1 aurora.enhanced_logical_replication=1 aurora.logical_replication_backup=0 aurora.logical_replication_globaldb=0 ``` 1. [选择或创建 Aurora PostgreSQL 数据库集群](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/CHAP_GettingStartedAurora.CreatingConnecting.Aurora.html)，并将步骤 1 创建的参数组与该数据库集群相关联。 1. 使用 [Aurora 和 AWS Secrets Manager的密码管理](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/rds-secrets-manager.html)，在 Amazon Aurora 集群上设置用户名和密码身份验证。您也可以通过创建 Secrets [Manager 密钥来创建 username/password ](https://docs.aws.amazon.com/secretsmanager/latest/userguide/create_secret.html)组合。 1. 如果您使用完整的初始快照功能，请创建一个 AWS KMS key 和一个 IAM 角色来将数据从 Amazon Aurora 导出到 Amazon S3。 IAM 角色还必须具有以下权限策略： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "ExportPolicy", "Effect": "Allow", "Action": [ "s3:PutObject*", "s3:ListBucket", "s3:GetObject*", "s3:DeleteObject*", "s3:GetBucketLocation" ], "Resource": [ "arn:aws:s3:::s3-bucket-used-in-pipeline", "arn:aws:s3:::s3-bucket-used-in-pipeline/*" ] } ] } ``` ------ 该角色还应拥有以下信任关系： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "export.rds.amazonaws.com" }, "Action": "sts:AssumeRole" } ] } ``` ------ 1. 选择或创建 OpenSearch 服务域或 OpenSearch 无服务器集合。有关更多信息，请参阅[创建 OpenSearch 服务域](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/createupdatedomains.html#createdomains)和[创建集合](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/serverless-manage.html#serverless-create)。 1. 将[基于资源的策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/ac.html#ac-types-resource)附加到域，或将[数据访问策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/serverless-data-access.html)附加到集合。这些访问策略允许 OpenSearch Ingestion 将数据从您的 Amazon Aurora 数据库集群写入您的域或集合。 ## 步骤 1：配置管道角色 完成 Amazon Aurora 管道先决条件设置后，[配置管道角色](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/pipeline-security-overview.html#pipeline-security-sink)，以便在管道配置中使用。还需为该角色添加以下针对 Amazon Aurora 源的权限： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "allowReadingFromS3Buckets", "Effect": "Allow", "Action": [ "s3:GetObject", "s3:DeleteObject", "s3:GetBucketLocation", "s3:ListBucket", "s3:PutObject" ], "Resource": [ "arn:aws:s3:::s3_bucket", "arn:aws:s3:::s3_bucket/*" ] }, { "Sid": "allowNetworkInterfacesActions", "Effect": "Allow", "Action": [ "ec2:AttachNetworkInterface", "ec2:CreateNetworkInterface", "ec2:CreateNetworkInterfacePermission", "ec2:DeleteNetworkInterface", "ec2:DeleteNetworkInterfacePermission", "ec2:DetachNetworkInterface", "ec2:DescribeNetworkInterfaces" ], "Resource": [ "arn:aws:ec2:*:111122223333:network-interface/*", "arn:aws:ec2:*:111122223333:subnet/*", "arn:aws:ec2:*:111122223333:security-group/*" ] }, { "Sid": "allowDescribeEC2", "Effect": "Allow", "Action": [ "ec2:Describe*" ], "Resource": "*" }, { "Sid": "allowTagCreation", "Effect": "Allow", "Action": [ "ec2:CreateTags" ], "Resource": "arn:aws:ec2:*:111122223333:network-interface/*", "Condition": { "StringEquals": { "aws:RequestTag/OSISManaged": "true" } } }, { "Sid": "AllowDescribeInstances", "Effect": "Allow", "Action": [ "rds:DescribeDBInstances" ], "Resource": [ "arn:aws:rds:us-east-2:111122223333:db:*" ] }, { "Sid": "AllowDescribeClusters", "Effect": "Allow", "Action": [ "rds:DescribeDBClusters" ], "Resource": [ "arn:aws:rds:us-east-2:111122223333:cluster:DB-id" ] }, { "Sid": "AllowSnapshots", "Effect": "Allow", "Action": [ "rds:DescribeDBClusterSnapshots", "rds:CreateDBClusterSnapshot", "rds:AddTagsToResource" ], "Resource": [ "arn:aws:rds:us-east-2:111122223333:cluster:DB-id", "arn:aws:rds:us-east-2:111122223333:cluster-snapshot:DB-id*" ] }, { "Sid": "AllowExport", "Effect": "Allow", "Action": [ "rds:StartExportTask" ], "Resource": [ "arn:aws:rds:us-east-2:111122223333:cluster:DB-id", "arn:aws:rds:us-east-2:111122223333:cluster-snapshot:DB-id*" ] }, { "Sid": "AllowDescribeExports", "Effect": "Allow", "Action": [ "rds:DescribeExportTasks" ], "Resource": "*", "Condition": { "StringEquals": { "aws:RequestedRegion": "us-east-2", "aws:ResourceAccount": "111122223333" } } }, { "Sid": "AllowAccessToKmsForExport", "Effect": "Allow", "Action": [ "kms:Decrypt", "kms:Encrypt", "kms:DescribeKey", "kms:RetireGrant", "kms:CreateGrant", "kms:ReEncrypt*", "kms:GenerateDataKey*" ], "Resource": [ "arn:aws:kms:us-east-2:111122223333:key/export-key-id" ] }, { "Sid": "AllowPassingExportRole", "Effect": "Allow", "Action": "iam:PassRole", "Resource": [ "arn:aws:iam::111122223333:role/export-role" ] }, { "Sid": "SecretsManagerReadAccess", "Effect": "Allow", "Action": [ "secretsmanager:GetSecretValue" ], "Resource": [ "arn:aws:secretsmanager:*:111122223333:secret:*" ] } ] } ``` ------ ## 步骤 2：创建管道 配置如下所示的 OpenSearch 摄取管道，该管道将 Aurora PostgreSQL 集群指定为源。 ``` version: "2" aurora-postgres-pipeline: source: rds: db_identifier: "cluster-id" engine: aurora-postgresql database: "database-name" tables: include: - "schema1.table1" - "schema2.table2" s3_bucket: "bucket-name" s3_region: "bucket-region" s3_prefix: "prefix-name" export: kms_key_id: "kms-key-id" iam_role_arn: "export-role-arn" stream: true aws: sts_role_arn: "arn:aws:iam::account-id:role/pipeline-role" region: "us-east-1" authentication: username: ${{aws_secrets:secret:username}} password: ${{aws_secrets:secret:password}} sink: - opensearch: hosts: ["https://search-mydomain.us-east-1.es.amazonaws.com"] index: "${getMetadata(\"table_name\")}" index_type: custom document_id: "${getMetadata(\"primary_key\")}" action: "${getMetadata(\"opensearch_action\")}" document_version: "${getMetadata(\"document_version\")}" document_version_type: "external" aws: sts_role_arn: "arn:aws:iam::account-id:role/pipeline-role" region: "us-east-1" extension: aws: secrets: secret: secret_id: "rds-secret-id" region: "us-east-1" sts_role_arn: "arn:aws:iam::account-id:role/pipeline-role" refresh_interval: PT1H ``` **注意** 您可以使用预先配置的 Amazon Aurora 蓝图，以创建此管道。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 要使用 Amazon Aurora 作为源，您需要为管道配置 VPC 访问权限。您选择的 VPC 应与 Amazon Aurora 源使用的 VPC 相同。然后选择一个或多个子网以及一个或多个 VPC 安全组。请注意，该管道需要访问 Aurora MySQL 数据库的网络权限，因此您还需确认 Aurora 集群已配置 VPC 安全组，允许来自管道 VPC 安全组的入站流量访问数据库端口。有关更多信息，请参阅[使用安全组控制访问权限](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/Overview.RDSSecurityGroups.html)。 如果您使用创建管道，则还必须将管道连接到 VPC，才能使用 Amazon Aurora 作为来源。 AWS 管理控制台 要执行此操作，请找到**网络配置**部分，选择**连接到 VPC**，然后从提供的任意一个默认选项中选择 CIDR，或者选择自己的 CIDR。您可以按照 [RFC 1918 当前最佳实践](https://datatracker.ietf.org/doc/html/rfc1918)中的定义，使用私有地址空间中的任何 CIDR。 要提供自定义 CIDR，请从下拉菜单中选择其他。为避免 OpenSearch Ingestion 和 Amazon Aurora 之间的 IP 地址冲突，请确保 Amazon Aurora VPC CIDR 与用于摄取的 CIDR 不同。 OpenSearch 有关更多信息，请参阅[为管道配置 VPC 访问权限](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/pipeline-security.html#pipeline-vpc-configure)。 ## 数据一致性 该管道通过持续轮询或接收来自 Amazon Aurora 集群的更改以及更新 OpenSearch索引中的相应文档来确保数据一致性。 OpenSearch Ingestion 支持 end-to-end确认，以确保数据的持久性。管道读取快照或流时，它会动态创建分区以进行并行处理。当管道在摄取 OpenSearch 域或集合中的所有记录后收到确认信息时，该管道会将该分区标记为已完成。如果要收录到 OpenSearch 无服务器搜索集合中，可以在管道中生成文档 ID。如果要采集到 OpenSearch Serverless 时间序列集合，请注意管道不会生成文档 ID，因此您必须在工作流接收器配置`document_id: "${getMetadata(\"primary_key\")}"`中省略文档 ID。 In OpenSearch gestion 管道还将传入的事件操作映射到相应的批量索引操作中，以帮助采集文档。这样可以保持数据的一致性，因此 Amazon Aurora 中的每一次数据更改都与中的 OpenSearch相应文档更改保持一致。 ## 映射数据类型 OpenSearch 摄取管道将 Aurora PostgreSQL 数据类型映射到适合 OpenSearch 服务域或集合使用的表示形式。如果中未定义映射模板 OpenSearch，则根据第一个发送的文档使用[动态映射 OpenSearch ](https://opensearch.org/docs/latest/field-types/#dynamic-mapping)自动确定字段类型。您还可以 OpenSearch 通过映射模板明确定义最适合您的字段类型。 下表列出了 Aurora PostgreSQL 数据类型和相应的字段类型。 OpenSearch OpenSearch 如果未定义显式映射，则 “*默认 OpenSearch 字段类型*” 列将显示相应的字段类型。在这种情况下，使用动态映射 OpenSearch 自动确定字段类型。“*推荐 OpenSearch 字段类型*” 列是在映射模板中明确指定的相应推荐字段类型。这些字段类型与 Aurora PostgreSQL 中的数据类型更加一致，通常可以启用中提供的更好的搜索功能。 OpenSearch | Aurora PostgreSQL 数据类型 | 默认 OpenSearch 字段类型 | 推荐的 OpenSearch 字段类型 | | --- | --- | --- | | smallint | 长整数 | short | | 整数 | 长整数 | integer | | bigint | 长整数 | 长整数 | | decimal | 文本 | 双精度或关键字 | | numeric[ (p, s) ] | 文本 | 双精度或关键字 | | real | float | 浮点数 | | double precision | 浮点数 | double | | smallserial | 长整数 | short | | Serial | 长整数 | 整数 | | bigserial | 长整数 | 长整数 | | money | object | object | | character varying(n) | 文本 | 文本 | | varchar(n) | 文本 | 文本 | | character(n) | 文本 | 文本 | | char(n) | 文本 | 文本 | | bpchar(n) | 文本 | 文本 | | bpchar | 文本 | 文本 | | 文本 | 文本 | 文本 | | enum | 文本 | 文本 | | bytea | 文本 | binary | | timestamp [ (p) ] [没有时区] | 长整数（以 epoch 毫秒为单位） | date | | 有时区的 timestamp [ (p) ] | 长整数（以 epoch 毫秒为单位） | date | | date | 长整数（以 epoch 毫秒为单位） | date | | time [ (p) ] [没有时区] | 长整数（以 epoch 毫秒为单位） | date | | 有时区的 time [ (p) ] | 长整数（以 epoch 毫秒为单位） | date | | interval [ fields ] [ (p) ] | 文本（ISO8601 格式） | 文本 | | 布尔值 | 布尔值 | 布尔值 | | point | 文本（采用 WKT 格式） | geo\$1shape | | 折线图 | 文本（采用 WKT 格式） | geo\$1shape | | lseg | 文本（采用 WKT 格式） | geo\$1shape | | box | 文本（采用 WKT 格式） | geo\$1shape | | path | 文本（采用 WKT 格式） | geo\$1shape | | 多边形 | 文本（采用 WKT 格式） | geo\$1shape | | circle | object | object | | cidr | 文本 | 文本 | | inet | 文本 | 文本 | | macaddr | 文本 | 文本 | | macaddr8 | 文本 | 文本 | | bit(n) | 长整数 | 字节、短整数、整数或长整数（具体取决于位数） | | bit varying(n) | 长整数 | 字节、短整数、整数或长整数（具体取决于位数） | | json | object | object | | jsonb | object | object | | jsonpath | 文本 | 文本 | 我们建议您在摄取管道中配置死信队列 (DLQ)。 OpenSearch 如果您已配置队列，S OpenSearch ervice 会将所有因动态映射失败而无法载入的失败文档发送到队列。 如果自动映射失败，则可以在管道配置中使用 `template_type` 和 `template_content` 来定义显式映射规则。或者，您可以在启动管道之前直接在搜索域或集合中创建映射模板。 ## 限制 在为 Aurora PostgreSQL 设置 OpenSearch 摄取管道时，请考虑以下限制： + 该集成仅支持每个管道一个 Aurora PostgreSQL 数据库。 + 该集成目前不支持跨区域数据摄取；您的 Amazon Aurora 集群和 OpenSearch 域必须处于相同状态。 AWS 区域 + 该集成目前不支持跨账户数据摄取；您的 Amazon Aurora 集群和 OpenSearch 摄取管道必须处于相同状态。 AWS 账户 + 确保 Amazon Aurora 集群已使用 AWS Secrets Manager启用身份验证，这是唯一受支持的身份验证机制。 + 无法更新现有的管道配置以从不同的数据库中提取不同 and/or 表的数据。要更新管道的数据库和/或表名称，则必须停止该管道，然后使用更新后的配置重新启动，或者创建新管道。 + 通常不支持数据定义语言（DDL）语句。以下情况下，数据一致性将无法得到保证： + 主键已更改 (add/delete/rename)。 + 表已被删除/截断。 + 列名或数据类型已发生更改。 + 如果要同步的 Aurora PostgreSQL 表未定义主键，则无法保证数据一致性。你需要在中定义自定义`document_id`选项 OpenSearch 和接收器配置才能同步 updates/deletes 到 OpenSearch。 + 支持的版本：Aurora PostgreSQL 版本 16.4 及更高版本。 ## 推荐 CloudWatch 警报 建议使用以下 CloudWatch 指标来监控您的摄取管道的性能。这些指标可帮助您确定处理的导出数据量、处理的流事件量、处理导出和流事件时的错误数以及写入目标的文档数量。您可以设置 CloudWatch 警报，以便在其中一个指标在指定时间内超过指定值时执行操作。 | 指标 | 说明 | | --- | --- | | pipeline-name.rds.Credentals已更改 | 该指标表示 AWS 密钥的轮换频率。 | | pipeline-name.rds。 executorRefreshErrors | 此指标表示无法刷新 AWS 密钥。 | | pipeline-name.rds。 exportRecordsTotal | 该指标指示从 Amazon Aurora 导出的记录数量。 | | pipeline-name.rds。 exportRecordsProcessed | 该指标表示 OpenSearch 摄取管道处理的记录数。 | | pipeline-name.rds。 exportRecordProcessing错误 | 该指标表示从 Amazon Aurora 集群读取数据时 OpenSearch 摄取管道中出现的处理错误数量。 | | pipeline-name.rds。 exportRecordsSuccess总计 | 该指标指示成功处理的导出记录总数。 | | pipeline-name.rds。 exportRecordsFailed总计 | 该指标指示处理失败的导出记录总数。 | | pipeline-name已收到.rds.bytes | 该指标表示 OpenSearch 摄取管道接收的总字节数。 | | pipeline-name.rds.bytes已处理 | 该指标表示 OpenSearch 摄取管道处理的总字节数。 | | pipeline-name.rds。 streamRecordsSuccess总计 | 该指标指示该流中成功处理的记录总数。 | | pipeline-name.rds。 streamRecordsFailed总计 | 该指标指示该流中处理失败的记录总数。 | # 在 Amaz OpenSearch on DynamoDB 上使用采集管道 您可以使用 D [ynamoDB](https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sources/dynamo-db/) 插件将表事件（例如创建、更新和删除）流式传输到亚马逊服务域和 OpenSearch 亚马逊 OpenSearch 无服务器集合。该管道采用变更数据捕获（CDC）技术实现大规模、低延迟的流式传输。 无论有没有完整初始快照，都可以处理 DynamoDB 数据。 + **有了完整快照**，DynamoDB [point-in-time 使用](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/PointInTimeRecovery.html)恢复 (PITR) 来创建备份并将其上传到 Amazon S3。 OpenSearch 然后，Ingestion 会将快照索引到一个或多个 OpenSearch索引中。为了保持一致性，管道会将所有 DynamoDB 更改与同步。 OpenSearch此选项要求同时启用 PITR 和 [DynamoDB Streams](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.CoreComponents.html#HowItWorks.CoreComponents.Streams)。 + **没有快照** — OpenSearch Ingestion 仅流式传输新的 DynamoDB 事件。如果您已有快照或者需要在无历史数据的情况下进行实时流式传输，则选择此选项。此选项要求仅启用 DynamoDB Streams。 *有关更多信息，请参阅《开发者指南》[中的 DynamoDB 零 ETL 与 OpenSearch 亚马逊服务的集成](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/OpenSearchIngestionForDynamoDB.html)。Amazon DynamoDB * **Topics** + [先决条件](#s3-prereqs) + [步骤 1：配置管道角色](#ddb-pipeline-role) + [步骤 2：创建管道](#ddb-pipeline) + [数据一致性](#ddb-pipeline-consistency) + [映射数据类型](#ddb-pipeline-mapping) + [限制](#ddb-pipeline-limitations) + [DynamoDB 的推荐 CloudWatch 警报](#ddb-pipeline-metrics) ## 先决条件 要设置管道，您必须有一个已启用 DynamoDB Streams 的 DynamoDB 表。您的流应使用 `NEW_IMAGE` 流视图类型。但是，`NEW_AND_OLD_IMAGES`如果这种流视图类型适合您的用例， OpenSearch Ingestion 管道也可以使用流式传输事件。 如果您使用的是快照，则还必须在表上启用 point-in-time恢复。有关更多信息，请参阅 *Amazon DynamoD* B 开发者[指南中的[创建表](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/WorkingWithTables.Basics.html#WorkingWithTables.Basics.CreateTable)、[启用 point-in-time恢复](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/PointInTimeRecovery_Howitworks.html#howitworks_enabling)和启用流](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html#Streams.Enabling)。 ## 步骤 1：配置管道角色 设置 DynamoDB 表后，[设置要在管道配置中使用的管道角色](pipeline-security-overview.md#pipeline-security-sink)，并在该角色中添加以下 DynamoDB 权限： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "allowRunExportJob", "Effect": "Allow", "Action": [ "dynamodb:DescribeTable", "dynamodb:DescribeContinuousBackups", "dynamodb:ExportTableToPointInTime" ], "Resource": [ "arn:aws:dynamodb:us-east-1:111122223333:table/my-table" ] }, { "Sid": "allowCheckExportjob", "Effect": "Allow", "Action": [ "dynamodb:DescribeExport" ], "Resource": [ "arn:aws:dynamodb:us-east-1:111122223333:table/my-table/export/*" ] }, { "Sid": "allowReadFromStream", "Effect": "Allow", "Action": [ "dynamodb:DescribeStream", "dynamodb:GetRecords", "dynamodb:GetShardIterator" ], "Resource": [ "arn:aws:dynamodb:us-east-1:111122223333:table/my-table/stream/*" ] }, { "Sid": "allowReadAndWriteToS3ForExport", "Effect": "Allow", "Action": [ "s3:GetObject", "s3:AbortMultipartUpload", "s3:PutObject", "s3:PutObjectAcl" ], "Resource": [ "arn:aws:s3:::amzn-s3-demo-bucket/export-folder/*" ] } ] } ``` ------ 您也可以使用 AWS 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 采集管道，将 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 蓝图，以创建此管道。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 ## 数据一致性 OpenSearch Ingestion 支持 end-to-end确认，以确保数据的持久性。管道读取快照或流时，它会动态创建分区以进行并行处理。当管道在摄取 OpenSearch 域或集合中的所有记录后收到确认信息时，该管道会将该分区标记为已完成。 如果要收录到 OpenSearch 无服务器*搜索*集合中，可以在管道中生成文档 ID。如果要采集到 OpenSearch 无服务器*时间序列*集合中，请注意该管道不会生成文档 ID。 In OpenSearch gestion 管道还将传入的事件操作映射到相应的批量索引操作中，以帮助采集文档。这样可以保持数据的一致性，因此 DynamoDB 中的每一次数据更改都与中的相应文档更改保持一致。 OpenSearch ## 映射数据类型 OpenSearch 服务将每个传入文档中的数据类型动态映射到 DynamoDB 中的相应数据类型。下表显示了 S OpenSearch ervice 如何自动映射各种数据类型。 | 数据类型 | OpenSearch | DynamoDB | | --- | --- | --- | | 数字 | OpenSearch 自动映射数值数据。如果该数字是整数，则将其 OpenSearch 映射为长值。如果数字是小数，则将其 OpenSearch 映射为浮点值。 OpenSearch 根据第一个发送的文档动态映射各种属性。如果您在 DynamoDB 中为同一属性混合了多种数据类型（例如整数和小数），则映射可能会失败。 例如，如果您的第一个文档的属性为整数，而后来的文档具有与小数相同的属性， OpenSearch 则无法采集第二个文档。在这些情况下，应提供一个显式的映射模板，如下所示： 
{
 "template": {
  "mappings": {
   "properties": {
    "MixedNumberAttribute": {
     "type": "float"
    }
   }
  }
 }
}
如果需要双精度，请使用字符串类型字段映射。不存在支持 38 位精度的等效数字类型 OpenSearch。 | DynamoDB 支持[数字](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.NamingRulesDataTypes.html#HowItWorks.DataTypes.Number)。 | | 数字集 | OpenSearch 自动将数字集映射到由长值或浮点值组成的数组中。与标量数字一样，这取决于摄取的第一个数字是整数还是小数。您可以像映射标量字符串一样提供数字集的映射。 | DynamoDB 支持表示[数字集](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.NamingRulesDataTypes.html#HowItWorks.DataTypes.SetTypes)的类型。 | | 字符串 | OpenSearch 自动将字符串值映射为文本。在某些情况下（例如枚举值），您可以映射到关键字类型。 以下示例说明如何将`PartType`名为的 DynamoDB 属性映射到关键字。 OpenSearch 
{
 "template": {
  "mappings": {
   "properties": {
    "PartType": {
     "type": "keyword"
    }
   }
  }
 }
}
| DynamoDB 支持[字符串](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.NamingRulesDataTypes.html#HowItWorks.DataTypes.String)。 | | 字符串集 | OpenSearch 自动将字符串集映射到字符串数组中。您可以像映射标量字符串一样提供字符串集的映射。 | DynamoDB 支持表示[字符串集](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.NamingRulesDataTypes.html#HowItWorks.DataTypes.SetTypes)的类型。 | | 二元 | OpenSearch 自动将二进制数据映射为文本。您可以提供映射以将它们写成二进制字段 OpenSearch。 以下示例说明如何将`ImageData`名为的 DynamoDB 属性映射到 OpenSearch 二进制字段。 
{
 "template": {
  "mappings": {
   "properties": {
    "ImageData": {
     "type": "binary"
    }
   }
  }
 }
}
| DynamoDB 支持[二进制类型属性](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.NamingRulesDataTypes.html#HowItWorks.DataTypes.Binary)。 | | 二进制集 | OpenSearch 自动将二进制集作为文本映射到二进制数据数组中。您可以像映射标量二进制一样提供数字集的映射。 | DynamoDB 支持表示[二进制值集](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.NamingRulesDataTypes.html#HowItWorks.DataTypes.SetTypes)的类型。 | | 布尔值 | OpenSearch 将 DynamoDB 布尔类型映射为 OpenSearch 布尔类型。 | DynamoDB 支持[布尔类型属性](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.NamingRulesDataTypes.html#HowItWorks.DataTypes.Boolean)。 | | Null | OpenSearch 可以采集 DynamoDB 空类型的文档。它将该值作为空值保存在文档中。此类型没有映射，并且此字段未编制索引或不可搜索。 如果对空类型使用相同的属性名称，然后更改为其他类型（例如字符串），则会为第一个非空值 OpenSearch 创建动态映射。后续值仍然可以是 DynamoDB 空值。 | DynamoDB 支持[空类型属性](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.NamingRulesDataTypes.html#HowItWorks.DataTypes.Null)。 | | Map | OpenSearch 将 DynamoDB 映射属性映射到嵌套字段。嵌套字段内也适用相同的映射。 以下示例将嵌套字段中的字符串映射到中的关键字类型 OpenSearch： 
{
 "template": {
  "mappings": {
   "properties": {
    "AdditionalDescriptions": {
     "properties": {
      "PartType": {
       "type": "keyword"
      }
     }
    }
   }
  }
 }
}
| DynamoDB 支持[映射类型属性](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.NamingRulesDataTypes.html#HowItWorks.DataTypes.Document.Map)。 | | 列表 | OpenSearch 根据列表中的内容，为 DynamoDB 列表提供了不同的结果。 当列表包含所有相同类型的标量类型（例如，所有字符串的列表）时，则将该列表作为 OpenSearch 该类型的数组提取。这适用于字符串、数字、布尔值和空类型。其中每种类型的限制与该类型标量的限制相同。 您还可以使用与映射相同的映射，为映射列表提供映射。 您无法提供混合类型的列表。 | DynamoDB 支持[列表类型属性](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.NamingRulesDataTypes.html#HowItWorks.DataTypes.Document.List)。 | | 设置 | OpenSearch 根据 DynamoDB 集合中的内容，为 DynamoDB 集提供不同的结果。 当一个集合包含所有相同类型的标量类型（例如，所有字符串的集合）时，则将该集合作为该类型的数组进行 OpenSearch 摄取。这适用于字符串、数字、布尔值和空类型。其中每种类型的限制与该类型标量的限制相同。 您还可以使用与映射相同的映射，为映射集合提供映射。 您无法提供混合类型的集合。 | DynamoDB 支持表示[集合](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/HowItWorks.NamingRulesDataTypes.html#HowItWorks.DataTypes.SetTypes)的类型。 | 我们建议您在摄取管道中配置死信队列 (DLQ)。 OpenSearch 如果您已配置队列，S OpenSearch ervice 会将所有因动态映射失败而无法载入的失败文档发送到队列。 如果自动映射失败，则可以在管道配置中使用 `template_type` 和 `template_content` 来定义显式映射规则。或者，您可以在启动管道之前直接在搜索域或集合中创建映射模板。 ## 限制 在为 DynamoDB 设置 OpenSearch 摄取管道时，请考虑以下限制： + OpenSearch 采集与 DynamoDB 的集成目前不支持跨区域接入。您的 DynamoDB 表 OpenSearch 和摄取管道必须处于相同的位置。 AWS 区域 + 您的 DynamoDB 表 OpenSearch 和摄取管道必须处于相同的位置。 AWS 账户 + 一个 OpenSearch 摄取管道仅支持一个 DynamoDB 表作为其来源。 + DynamoDB Streams 仅在日志中存储最多 24 小时的数据。如果从大型表的初始快照中摄取数据需要 24 小时或更长时间，则会丢失一些初始数据。为了缓解这种数据丢失，请估计表的大小并配置适当的 OpenSearch 摄取管道计算单元。 ## DynamoDB 的推荐 CloudWatch 警报 建议使用以下 CloudWatch 指标来监控您的摄取管道的性能。这些指标可能有助您确定处理的导出数据量、处理的流事件量、处理导出和流事件时的错误数以及写入目标的文档数量。您可以设置 CloudWatch 警报，以便在其中一个指标在指定时间内超过指定值时执行操作。 | 指标 | 说明 | | --- |--- | | dynamodb-pipeline.BlockingBuffer.bufferUsage.value | 指示正在使用的缓冲区数量。 | | dynamodb-pipeline.dynamodb.activeExportS3ObjectConsumers.value | 显示正在积极处理待导出 OCUs 的 Amazon S3 对象的总数。 | | dynamodb-pipeline.dynamodb.bytesProcessed.count | 处理的 DynamoDB 源字节数。 | | dynamodb-pipeline.dynamodb.changeEventsProcessed.count | 处理的 DynamoDB 流更改事件数量。 | | dynamodb-pipeline.dynamodb.changeEventsProcessingErrors.count | 处理的 DynamoDB 更改事件中的错误数。 | | dynamodb-pipeline.dynamodb.exportJobFailure.count | 失败的导出任务提交尝试次数。 | | dynamodb-pipeline.dynamodb.exportJobSuccess.count | 已成功提交的导出任务数量。 | | dynamodb-pipeline.dynamodb.exportRecordsProcessed.count | 处理的导出记录总数。 | | dynamodb-pipeline.dynamodb.exportRecordsTotal.count | 从 DynamoDB 导出的记录总数，这对于跟踪数据导出量至关重要。 | | dynamodb-pipeline.dynamodb.exportS3ObjectsProcessed.count | 从 Amazon S3 成功处理的导出数据文件总数。 | | dynamodb-pipeline.opensearch.bulkBadRequestErrors.count | 批量请求期间由于请求格式错误而导致的错误计数。 | | dynamodb-pipeline.opensearch.bulkRequestLatency.avg | 向发出的批量写入请求的平均延迟 OpenSearch。 | | dynamodb-pipeline.opensearch.bulkRequestNotFoundErrors.count | 由于找不到目标数据而失败的批量请求数。 | | dynamodb-pipeline.opensearch.bulkRequestNumberOfRetries.count | 采集管道为写 OpenSearch 入集群而进行的重试次数。 OpenSearch | | dynamodb-pipeline.opensearch.bulkRequestSizeBytes.sum | 向发出的所有批量请求的总大小（以字节为单位） OpenSearch。 | | dynamodb-pipeline.opensearch.documentErrors.count | 向发送文档时出现的错误数 OpenSearch。导致错误的文件将发送给 DLQ。 | | dynamodb-pipeline.opensearch.documentsSuccess.count | 成功写入 OpenSearch 集群或集合的文档数。 | | dynamodb-pipeline.opensearch.documentsSuccessFirstAttempt.count | 第一次尝试成功编入索引 OpenSearch 的文档数。 | | `dynamodb-pipeline.opensearch.documentsVersionConflictErrors.count` | 处理过程中由于文档版本冲突而导致的错误计数。 | | `dynamodb-pipeline.opensearch.PipelineLatency.avg` | OpenSearch Ingestion 管道通过从源读取数据到写入目标来处理数据的平均延迟。 | | dynamodb-pipeline.opensearch.PipelineLatency.max | 通过从源读取数据到写入目标来处理数据的 OpenSearch Ingestion 管道的最大延迟。 | | dynamodb-pipeline.opensearch.recordsIn.count | 成功摄入 OpenSearch的记录数。该指标对于跟踪正在处理和存储的数据量至关重要。 | | dynamodb-pipeline.opensearch.s3.dlqS3RecordsFailed.count | 未能写入 DLQ 的记录数。 | | dynamodb-pipeline.opensearch.s3.dlqS3RecordsSuccess.count | 写入 DLQ 的记录数。 | | dynamodb-pipeline.opensearch.s3.dlqS3RequestLatency.count | Amazon S3 死信队列请求的延迟测量次数。 | | dynamodb-pipeline.opensearch.s3.dlqS3RequestLatency.sum | 向 Amazon S3 死信队列发出的所有请求的总延迟 | | dynamodb-pipeline.opensearch.s3.dlqS3RequestSizeBytes.sum | 向 Amazon S3 死信队列发出的所有请求的总大小（以字节为单位）。 | | dynamodb-pipeline.recordsProcessed.count | 管道中处理的记录总数，这是衡量整体吞吐量的关键指标。 | | dynamodb.changeEventsProcessed.count | 没有从 DynamoDB 流收集任何记录。这可能是由于桌上没有活动、正在进行导出或访问 DynamoDB 直播时出现问题。 | | `dynamodb.exportJobFailure.count` | 尝试触发向 S3 的导出操作失败。 | | `dynamodb-pipeline.opensearch.bulkRequestInvalidInputErrors.count` | OpenSearch 由于输入无效而导致的批量请求错误计数，这对于监控数据质量和操作问题至关重要。 | | opensearch.EndToEndLatency.avg | 端到端延迟高于从 DynamoDB 数据流读取所需的延迟。这可能是由于 OpenSearch 集群规模不足，或者管道 OCU 的最大容量对于 DynamoDB 表上的 WCU 吞吐量来说过低。这种端到端的延迟在导出后会很高，而且随着时间的推移，它会赶上最新的 DynamoDB 流，因此会随着时间的推移而降低。 | # 在 Amaz OpenSearch on DocumentDB 中使用采集管道 您可以使用 D [ocumentDB](https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sources/documentdb/) 插件将文档更改（例如创建、更新和删除）流式传输到亚马逊 OpenSearch 服务。该管道支持变更数据捕获（CDC）（如可用）或 API 轮询，以实现大规模、低延迟的流式传输。 无论有没有完整初始快照，都可以处理数据。完整快照会捕获整个 Amazon DocumentDB 集合，并将其上传至 Amazon S3。然后，管道将数据发送到一个或多个 OpenSearch 索引。摄取快照后，管道会同步正在进行的更改以保持一致性，并最终赶上近乎实时的更新。 如果您已有来自其他源的完整快照，或者只需处理新事件，则可以在没有快照的情况下进行流式传输。在此情况下，管道会直接读取 Amazon DocumentDB 更改流，无需初始批量加载。 如果您启用流，则必须在 Amazon DocumentDB 集合上[启用更改流](https://docs.aws.amazon.com/documentdb/latest/developerguide/change_streams.html#change_streams-enabling)。然而，如果您只执行满载或导出操作，则不需要更改流。 ## 先决条件 在创建 OpenSearch Ingestion 管道之前，请执行以下步骤： 1. 按照《Amazon DocumentDB 开发者指南》中 [Create an Amazon DocumentDB cluster](https://docs.aws.amazon.com/documentdb/latest/developerguide/get-started-guide.html#cloud9-cluster)** 说明的步骤操作，创建一个具有数据读取权限的 Amazon DocumentDB 集群。如果使用 CDC 基础设施，将 Amazon DocumentDB 集群配置为发布更改流。 1. 在 Amazon DocumentDB 集群上启用 TLS。 1. 设置私有地址空间的 VPC CIDR 以用于 OpenSearch 接入。 1. 使用在您的亚马逊文档数据库集群上设置身份验证。 AWS Secrets Manager按照 [Automatically Rotating Passwords for Amazon DocumentDB](https://docs.aws.amazon.com/documentdb/latest/developerguide/security.managing-users.html#security.managing-users-rotating-passwords) 中的步骤启用密钥轮换。有关更多信息，请参阅 [Database access using Role-Based Access Control](https://docs.aws.amazon.com/documentdb/latest/developerguide/role_based_access_control.html) 和 [Security in Amazon DocumentDB](https://docs.aws.amazon.com/documentdb/latest/developerguide/security.html)。 1. 如果您使用更改流订阅 Amazon DocumentDB 集合中的数据更改，请使用 `change_stream_log_retention_duration` 参数将保留期延长至最多 7 天，以避免数据丢失。默认情况下，更改流事件将在事件录制后存储 3 小时，这对于大型集合来说是不够的。要修改更改流的保留期，请参阅 [Modifying the change stream log retention duration](https://docs.aws.amazon.com/documentdb/latest/developerguide/change_streams.html#change_streams-modifying_log_retention)。 1. 创建 OpenSearch 服务域或 OpenSearch 无服务器集合。有关更多信息，请参阅[创建 OpenSearch 服务域](createupdatedomains.md#createdomains)和[创建集合](serverless-create.md)。 1. 将[基于资源的策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/ac.html#ac-types-resource)附加到域，或将[数据访问策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/serverless-data-access.html)附加到集合。这些访问策略允许 OpenSearch Ingestion 将数据从您的 Amazon DocumentDB 集群写入您的域名或集合。 以下示例域访问策略允许您在下一步中创建的管道角色将数据写入域。确保使用自身 ARN 更新 `resource`。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "AWS": "arn:aws:iam::444455556666:role/pipeline-role" }, "Action": [ "es:DescribeDomain", "es:ESHttp*" ], "Resource": [ "arn:aws:es:us-east-1:111122223333:domain/domain-name" ] } ] } ``` ------ 要创建具有向该集合或域写入数据的正确权限的 IAM 角色，请参阅 [在 Amazon OpenSearch Ingestion 中设置角色和用户](pipeline-security-overview.md)。 ## 步骤 1：配置管道角色 完成 Amazon DocumentDB 管道先决条件设置后，[配置管道角色](pipeline-security-overview.md#pipeline-security-sink)以在管道配置中使用，并为该角色添加以下 Amazon DocumentDB 权限： ------ #### [ JSON ] **** ``` { "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:us-east-1:111122223333:secret:secret-name" ] }, { "Effect": "Allow", "Action": [ "ec2:AttachNetworkInterface", "ec2:CreateNetworkInterface", "ec2:CreateNetworkInterfacePermission", "ec2:DeleteNetworkInterface", "ec2:DeleteNetworkInterfacePermission", "ec2:DetachNetworkInterface", "ec2:DescribeNetworkInterfaces" ], "Resource": [ "arn:aws:ec2:*:111122223333:network-interface/*", "arn:aws:ec2:*:111122223333:subnet/*", "arn:aws:ec2:*:111122223333: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 摄取管道的 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: 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" refresh_interval: PT1H ``` 您可以使用预先配置的 Amazon DocumentDB 蓝图，以创建此管道。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 如果您使用创建管道，则还必须将管道连接到 VPC，才能使用 Amazon DocumentDB 作为来源。 AWS 管理控制台 要执行此操作，请找到**源网络选项**部分，选择**连接到 VPC** 复选框，然后从提供的任意一个默认选项中选择 CIDR。您可以按照 [RFC 1918 当前最佳实践](https://datatracker.ietf.org/doc/html/rfc1918)中的定义，使用私有地址空间中的任何 CIDR。 要提供自定义 CIDR，请从下拉菜单中选择**其他**。为避免 OpenSearch Ingestion 和 Amazon DocumentDB 之间的 IP 地址冲突，请确保亚马逊 DocumentDB VPC CIDR 与用于摄取的 CIDR 不同。 OpenSearch 有关更多信息，请参阅[为管道配置 VPC 访问权限](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/pipeline-security.html#pipeline-vpc-configure)。 ## 数据一致性 该管道通过持续轮询或接收来自 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 支持[整型](https://docs.aws.amazon.com/documentdb/latest/developerguide/mongo-apis.html#mongo-apis-data-types)。 | | 长整型 | OpenSearch 自动将 Amazon DocumentDB 的长整数值映射到 OpenSearch多头。 OpenSearch 根据第一个发送的文档动态映射该字段。如果 Amazon DocumentDB 中的同一属性混合了多种数据类型，则自动映射可能会失败。 例如，如果您的第一个文档的属性很长，而后来的文档具有与整数相同的属性， OpenSearch 则无法收录第二个文档。对于这些情况，应提供选择最灵活数值类型的显式映射模板，如下所示： 
{
 "template": {
  "mappings": {
   "properties": {
    "MixedNumberField": {
     "type": "float"
    }
   }
  }
 }
}
| Amazon DocumentDB 支持[长整型](https://docs.aws.amazon.com/documentdb/latest/developerguide/mongo-apis.html#mongo-apis-data-types)。 | | 字符串 | OpenSearch 自动将字符串值映射为文本。在某些情况下（例如枚举值），您可以映射到关键字类型。 以下示例说明如何将名为的 Amazon DocumentDB 属性映射`PartType`到关键词。 OpenSearch 
{
 "template": {
  "mappings": {
   "properties": {
    "PartType": {
     "type": "keyword"
    }
   }
  }
 }
}
| Amazon DocumentDB 支持[字符串](https://docs.aws.amazon.com/documentdb/latest/developerguide/mongo-apis.html#mongo-apis-data-types)。 | | 双精度 | OpenSearch 自动将 Amazon DocumentDB 双精度值映射到双精度值。 OpenSearch OpenSearch 根据第一个发送的文档动态映射该字段。如果 Amazon DocumentDB 中的同一属性混合了多种数据类型，则自动映射可能会失败。 例如，如果您的第一个文档的属性很长，而后来的文档具有与整数相同的属性， OpenSearch 则无法收录第二个文档。对于这些情况，应提供选择最灵活数值类型的显式映射模板，如下所示： 
{
 "template": {
  "mappings": {
   "properties": {
    "MixedNumberField": {
     "type": "float"
    }
   }
  }
 }
}
| Amazon DocumentDB 支持[双精度浮点数](https://docs.aws.amazon.com/documentdb/latest/developerguide/mongo-apis.html#mongo-apis-data-types)。 | | 日期 | 默认情况下，日期映射到中的整数 OpenSearch。您可以定义自定义映射模板以将日期映射到 OpenSearch 日期。 
{
 "template": {
  "mappings": {
   "properties": {
    "myDateField": {
     "type": "date",
     "format": "epoch_second"
    }
   }
  }
 }
}
| Amazon DocumentDB 支持[日期](https://docs.aws.amazon.com/documentdb/latest/developerguide/mongo-apis.html#mongo-apis-data-types)。 | | Timestamp | 默认情况下，时间戳映射到中的整数。 OpenSearch您可以定义自定义映射模板以将日期映射到 OpenSearch日期。 
{
 "template": {
  "mappings": {
   "properties": {
    "myTimestampField": {
     "type": "date",
     "format": "epoch_second"
    }
   }
  }
 }
}
| Amazon DocumentDB 支持[时间戳](https://docs.aws.amazon.com/documentdb/latest/developerguide/mongo-apis.html#mongo-apis-data-types)。 | | 布尔值 | OpenSearch 将 Amazon DocumentDB 布尔类型映射为 OpenSearch 布尔类型。 | Amazon DocumentDB 支持[布尔类型属性](https://docs.aws.amazon.com/documentdb/latest/developerguide/mongo-apis.html#mongo-apis-data-types)。 | | 十进制 | OpenSearch 将 Amazon DocumentDB 映射属性映射到嵌套字段。嵌套字段内也适用相同的映射。 以下示例将嵌套字段中的字符串映射到中的关键字类型 OpenSearch： 
{
 "template": {
  "mappings": {
   "properties": {
    "myDecimalField": {
     "type": "double"
    }
   }
  }
 }
}
使用此自定义映射，您可以对该字段进行双精度查询和聚合。原始值保留了 OpenSearch文档`_source`属性的全部精度。如果没有此映射，则默认 OpenSearch 使用文本。 | Amazon DocumentDB 支持[小数](https://docs.aws.amazon.com/documentdb/latest/developerguide/mongo-apis.html#mongo-apis-data-types)。 | | 正则表达式 | 正则表达式类型会创建嵌套字段。这包括 .pattern 和 .options。 | Amazon DocumentDB 支持[正则表达式](https://docs.aws.amazon.com/documentdb/latest/developerguide/mongo-apis.html#mongo-apis-data-types)。 | | 二进制数据 | OpenSearch 自动将 Amazon DocumentDB 二进制数据映射到 OpenSearch文本。您可以提供映射以将它们写成二进制字段 OpenSearch。 以下示例说明如何将名为的 Amazon DocumentDB 字段映射`imageData`到 OpenSearch 二进制字段。 
{
 "template": {
  "mappings": {
   "properties": {
    "imageData": {
     "type": "binary"
    }
   }
  }
 }
}
| Amazon DocumentDB 支持[二进制数据字段](https://docs.aws.amazon.com/documentdb/latest/developerguide/mongo-apis.html#mongo-apis-data-types)。 | | ObjectId | 具有 ObjectID 类型的字段会映射到 OpenSearch 文本字段。该值将是 objectId 的字符串表示形式。 | Amazon DocumentDB 支持 [objectIds](https://docs.aws.amazon.com/documentdb/latest/developerguide/mongo-apis.html#mongo-apis-data-types)。 | | Null | OpenSearch 可以收录 Amazon DocumentDB 空类型的文档。它将该值作为空值保存在文档中。此类型没有映射，并且此字段未编制索引或不可搜索。 如果对空类型使用相同的属性名称，然后更改为其他类型（例如字符串），则会为第一个非空值 OpenSearch 创建动态映射。后续值仍然可以是 Amazon DocumentDB 空值。 | Amazon DocumentDB 支持[空值类型字段](https://docs.aws.amazon.com/documentdb/latest/developerguide/mongo-apis.html#mongo-apis-data-types)。 | | 未定义 | OpenSearch 可以收录 Amazon DocumentDB 未定义类型的文档。它将该值作为空值保存在文档中。此类型没有映射，并且此字段未编制索引或不可搜索。 如果对未定义的类型使用相同的字段名称，然后更改为其他类型（例如字符串），则会为第一个未定义的值 OpenSearch 创建动态映射。后续值仍然可以是 Amazon DocumentDB 未定义值。 | Amazon DocumentDB 支持[未定义类型字段](https://docs.aws.amazon.com/documentdb/latest/developerguide/mongo-apis.html#mongo-apis-data-types)。 | | MinKey | OpenSearch 可以收录亚马逊 DocumentDB minKey 类型的文档。它将该值作为空值保存在文档中。此类型没有映射，并且此字段未编制索引或不可搜索。 如果对 minKey 类型使用相同的字段名称，然后更改为其他类型（例如字符串），则会为第一个非 MinKey 值 OpenSearch 创建动态映射。后续值仍然可以是 Amazon DocumentDB minKey 值。 | Amazon DocumentDB 支持 [minKey 类型字段](https://docs.aws.amazon.com/documentdb/latest/developerguide/mongo-apis.html#mongo-apis-data-types)。 | | MaxKey | OpenSearch 可以收录亚马逊 DocumentDB MaxKey 类型的文档。它将该值作为空值保存在文档中。此类型没有映射，并且此字段未编制索引或不可搜索。 如果对 MaxKey 类型使用相同的字段名称，然后更改为其他类型（例如字符串），则会为第一个非 MaxKey 值 OpenSearch 创建动态映射。后续值仍然可以是 Amazon DocumentDB maxKey 值。 | Amazon DocumentDB 支持 [maxKey 类型字段](https://docs.aws.amazon.com/documentdb/latest/developerguide/mongo-apis.html#mongo-apis-data-types)。 | 我们建议您在摄取管道中配置死信队列 (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 警报 为了获得最佳性能，我们建议您在创建 OpenSearch 采集管道时使用以下 CloudWatch 警报来访问作为来源的 Amazon DocumentDB 集群。 | CloudWatch 警报 | 说明 | | --- | --- | | .docucmentdb.凭证已更改 | 该指标表示 AWS 密钥的轮换频率。 | | .docmentdb。 executorRefreshErrors | 该指标指示刷新 AWS 密钥失败。 | | .docmentdb。 exportRecordsTotal | 该指标指示从 Amazon DocumentDB 导出的记录数量。 | | .docmentdb。 exportRecordsProcessed | 该指标表示 OpenSearch 摄取管道处理的记录数。 | | .docmentdb。 exportRecordProcessing错误 | 该指标表示从 Amazon DocumentDB 集群读取数据时 OpenSearch 摄取管道中出现的处理错误数量。 | | .docmentdb。 exportRecordsSuccess总计 | 该指标指示成功处理的导出记录总数。 | | .docmentdb。 exportRecordsFailed总计 | 该指标指示处理失败的导出记录总数。 | | .docucmentdb.bytes 已接收 | 该指标表示 OpenSearch 摄取管道接收的总字节数。 | | .docucmentdb.bytes已处理 | 该指标表示 OpenSearch 摄取管道处理的总字节数。 | | .docmentdb。 exportPartitionQuery总计 | 该指标指示导出分区总数。 | | .docmentdb。 streamRecordsSuccess总计 | 该指标指示该流中成功处理的记录总数。 | | .docmentdb。 streamRecordsFailed总计 | 该指标指示该流中处理失败的记录总数。 | # 在 Confluent Cl OpenSearch oud Kafka 中使用采集管道 您可以使用 OpenSearch 摄取管道将数据从 Confluent Cloud Kafka 集群流式传输到 OpenSearch 亚马逊服务域和无服务器集合。 OpenSearch OpenSearch Ingestion 支持公用和私有网络配置，用于将数据从 Confluent Cloud Kafka 集群流式传输到由服务或无服务器管理的域或集合。 OpenSearch OpenSearch ## 连接到 Confluent Cloud 公有 Kafka 集群 您可以使用 OpenSearch 摄取管道从具有公共配置的 Confluent Cloud Kafka 集群迁移数据，这意味着可以公开解析域 DNS 名称。为此，请设置一个 OpenSearch 采集管道，将 Confluent Cloud 公共 Kafka 集群作为源，将 OpenSearch 服务或 OpenSearch 无服务器作为目标。这将处理您从自行管理的源集群到托 AWS管目标域或集合的流式数据。 ### 先决条件 在创建 OpenSearch 摄取管道之前，请执行以下步骤： 1. 创建充当源的 Confluent Cloud Kafka 集群。集群应包含您要采集到 S OpenSearch ervice 中的数据。 1. 创建要将数据迁移到的 OpenSearch 服务域或 OpenSearch 无服务器集合。有关更多信息，请参阅[创建 OpenSearch 服务域](createupdatedomains.md#createdomains)和[创建集合](serverless-create.md)。 1. 使用 AWS Secrets Manager在 Confluent Cloud Kafka 集群上设置身份验证。按照 [Rotate AWS Secrets Manager secrets](https://docs.aws.amazon.com/secretsmanager/latest/userguide/rotating-secrets.html) 中的步骤启用密钥轮换。 1. 将[基于资源的策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/ac.html#ac-types-resource)附加到域，或将[数据访问策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/serverless-data-access.html)附加到集合。这些访问策略允许 OpenSearch Ingestion 将数据从您的自管理集群写入您的域或集合。 以下示例域访问策略允许您在下一步中创建的管道角色将数据写入域。确保使用自身 ARN 更新 `resource`。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "AWS": "arn:aws:iam::444455556666:role/pipeline-role" }, "Action": [ "es:DescribeDomain", "es:ESHttp*" ], "Resource": [ "arn:aws:es:us-east-1:111122223333:domain/domain-name" ] } ] } ``` ------ 要创建具有向该集合或域写入数据的正确权限的 IAM 角色，请参阅 [在 Amazon OpenSearch Ingestion 中设置角色和用户](pipeline-security-overview.md)。 ### 步骤 1：配置管道角色 设置 Confluent Cloud Kafka 集群管道先决条件后，[配置要在管道配置中使用的管道角色](pipeline-security-overview.md#pipeline-security-sink)，添加写入 OpenSearch 服务域或 OpenSearch 无服务器集合的权限，以及从 Secrets Manager 读取密钥的权限。 管理网络接口需要以下权限： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "ec2:AttachNetworkInterface", "ec2:CreateNetworkInterface", "ec2:CreateNetworkInterfacePermission", "ec2:DeleteNetworkInterface", "ec2:DeleteNetworkInterfacePermission", "ec2:DetachNetworkInterface", "ec2:DescribeNetworkInterfaces" ], "Resource": [ "arn:aws:ec2:us-east-1:111122223333:network-interface/*", "arn:aws:ec2:us-east-1:111122223333:subnet/*", "arn:aws:ec2:us-east-1:111122223333:security-group/*" ] }, { "Effect": "Allow", "Action": [ "ec2:DescribeDhcpOptions", "ec2:DescribeRouteTables", "ec2:DescribeSecurityGroups", "ec2:DescribeSubnets", "ec2:DescribeVpcs", "ec2:Describe*" ], "Resource": "arn:aws:ec2:us-east-1:111122223333:subnet/*" }, { "Effect": "Allow", "Action": [ "ec2:CreateTags" ], "Resource": "arn:aws:ec2:us-east-1:111122223333:network-interface/*", "Condition": { "StringEquals": { "aws:RequestTag/OSISManaged": "true" } } } ] } ``` ------ 以下是从 AWS Secrets Manager 服务中读取机密所需的权限： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "SecretsManagerReadAccess", "Effect": "Allow", "Action": ["secretsmanager:GetSecretValue"], "Resource": ["arn:aws:secretsmanager:us-east-1:111122223333:secret:,secret-name"] } ] } ``` ------ 写入亚马逊 OpenSearch 服务域需要以下权限： ``` { "Statement": [ { "Effect": "Allow", "Principal": { "AWS": "arn:aws:iam::account-id:role/pipeline-role" }, "Action": ["es:DescribeDomain", "es:ESHttp*"], "Resource": "arn:aws:es:region:account-id:domain/domain-name/*" } ] } ``` ### 步骤 2：创建管道 然后，你可以配置如下所示的 OpenSearch 摄取管道，它将你的 Confluent Cloud Kafka 指定为来源。 您可以指定多个 OpenSearch 服务域作为数据的目的地。此功能允许有条件地路由或将传入数据复制到多个 OpenSearch 服务域。 您还可以将数据从源 Confluent Kafka 集群迁移到无服务器 V OpenSearch PC 集合。务必要在管道配置中提供网络访问策略。您可以使用 Confluent 架构注册表来定义 Confluent 架构。 ``` version: "2" kafka-pipeline: source: kafka: encryption: type: "ssl" topics: - name: "topic-name" group_id: "group-id" bootstrap_servers: - "bootstrap-server.us-east-1.aws.private.confluent.cloud:9092" authentication: sasl: plain: username: ${aws_secrets:confluent-kafka-secret:username} password: ${aws_secrets:confluent-kafka-secret:password} schema: type: confluent registry_url: https://my-registry.us-east-1.aws.confluent.cloud api_key: "${{aws_secrets:schema-secret:schema_registry_api_key}}" api_secret: "${{aws_secrets:schema-secret:schema_registry_api_secret}}" basic_auth_credentials_source: "USER_INFO" sink: - opensearch: hosts: ["https://search-mydomain.us-east-1.es.amazonaws.com"] aws: region: "us-east-1" aws: secrets: confluent-kafka-secret: secret_id: "my-kafka-secret" region: "us-east-1" schema-secret: secret_id: "my-self-managed-kafka-schema" region: "us-east-1" ``` 您可以使用预先配置的蓝图，以创建此管道。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 ### 连接到 VPC 中的 Confluent Cloud Kafka 集群 您还可以使用 OpenSearch 摄取管道从 VPC 中运行的 Confluent Cloud Kafka 集群迁移数据。为此，请设置一个 OpenSearch 采集管道，将 Confluent Cloud Kafka 集群作为源，将 OpenSearch 服务或 OpenSearch 无服务器作为目标。这将处理从 Confluent Cloud Kafka 源集群到 AWS托管式目标域或集合的流式数据传输。 OpenSearch Ingestion 支持在 Confluent 中以所有支持的网络模式配置的 Confluent Cloud Kafka 集群。 OpenSearch Ingestion 支持以下网络配置模式作为来源： + AWS VPC 对等连接 + AWS PrivateLink 适用于专用集群 + AWS PrivateLink 适用于企业集群 + AWS Transit Gateway #### 先决条件 在创建 OpenSearch 摄取管道之前，请执行以下步骤： 1. 使用 VPC 网络配置创建 Confluent Cloud Kafka 集群，其中包含您要提取到服务中的数据。 OpenSearch 1. 创建要将数据迁移到的 OpenSearch 服务域或 OpenSearch 无服务器集合。有关更多信息，请参阅 [创建 OpenSearch 服务域](createupdatedomains.md#createdomains) 和 [创建集合](serverless-create.md)。 1. 使用 AWS Secrets Manager在 Confluent Cloud Kafka 集群上设置身份验证。按照 [Rotate AWS Secrets Manager secrets](https://docs.aws.amazon.com/secretsmanager/latest/userguide/rotating-secrets.html) 中的步骤启用密钥轮换。 1. 获取有权访问 Confluent Cloud Kafka 集群的 VPC 的 ID。选择 OpenSearch Ingestion 要使用的 VPC CIDR。 **注意** 如果您使用创建管道，则还必须将您的 OpenSearch 摄取管道连接到您的 VPC 才能使用 Confluent Cloud Kafka 集群。 AWS 管理控制台 要执行此操作，请找到**网络配置**部分，选择**连接到 VPC** 复选框，然后从提供的任意一个默认选项中选择 CIDR，或者选择自己的 CIDR。您可以按照 [RFC 1918 当前最佳实践](https://datatracker.ietf.org/doc/html/rfc1918)中的定义，使用私有地址空间中的任何 CIDR。 要提供自定义 CIDR，请从下拉菜单中选择**其他**。为避免接入和自行管理之间的 IP 地址冲突 OpenSearch，请确保自管理 VP OpenSearch C CIDR 与用于 OpenSearch 摄取的 CIDR 不同。 OpenSearch 1. 将[基于资源的策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/ac.html#ac-types-resource)附加到域，或将[数据访问策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/serverless-data-access.html)附加到集合。这些访问策略允许 OpenSearch Ingestion 将数据从您的自管理集群写入您的域或集合。 **注意** 如果使用 AWS PrivateLink 连接 Confluent Cloud Kafka，则需要配置 [VPC DHCP 选项](https://docs.aws.amazon.com/vpc/latest/userguide/VPC_DHCP_Options.html)。应启用 *DNS 主机名*和 *DNS 解析*。 具体而言，使用以下选项集值： **企业集群：** ``` domain-name: aws.private.confluent.cloud domain-name-servers: AmazonProvidedDNS ``` **专用集群：** ``` domain-name: aws.confluent.cloud domain-name-servers: AmazonProvidedDNS ``` 此更改可确保 Confluent PrivateLink 终端节点的 DNS 解析在 VPC 内正常运行。 以下示例域访问策略允许您在下一步中创建的管道角色将数据写入域。确保使用自身 ARN 更新 `resource`。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "AWS": "arn:aws:iam::444455556666:role/pipeline-role" }, "Action": [ "es:DescribeDomain", "es:ESHttp*" ], "Resource": [ "arn:aws:es:us-east-1:111122223333:domain/domain-name" ] } ] } ``` ------ 要创建具有向该集合或域写入数据的正确权限的 IAM 角色，请参阅 [在 Amazon OpenSearch Ingestion 中设置角色和用户](pipeline-security-overview.md)。 #### 步骤 1：配置管道角色 完成管道先决条件设置后，[配置管道角色](pipeline-security-overview.md#pipeline-security-sink)以在管道配置中使用，并为该角色添加以下权限： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "SecretsManagerReadAccess", "Effect": "Allow", "Action": [ "secretsmanager:GetSecretValue" ], "Resource": ["arn:aws:secretsmanager:us-east-1:111122223333:secret:secret-name"] }, { "Effect": "Allow", "Action": [ "ec2:AttachNetworkInterface", "ec2:CreateNetworkInterface", "ec2:CreateNetworkInterfacePermission", "ec2:DeleteNetworkInterface", "ec2:DeleteNetworkInterfacePermission", "ec2:DetachNetworkInterface", "ec2:DescribeNetworkInterfaces" ], "Resource": [ "arn:aws:ec2:*:*:network-interface/*", "arn:aws:ec2:*:*:subnet/*", "arn:aws:ec2:*:*: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 摄取管道的 IAM 角色提供上述 Amazon EC2 权限，因为管道使用这些权限在您的 VPC 中创建和删除网络接口。该管道只能通过此网络接口访问 Kafka 集群。 #### 步骤 2：创建管道 然后，您可以配置如下所示的 OpenSearch 采集管道，将 Kafka 指定为来源。 您可以指定多个 OpenSearch 服务域作为数据的目的地。此功能允许有条件地路由或将传入数据复制到多个 OpenSearch 服务域。 您还可以将数据从源 Confluent Kafka 集群迁移到无服务器 V OpenSearch PC 集合。务必要在管道配置中提供网络访问策略。您可以使用 Confluent 架构注册表来定义 Confluent 架构。 ``` version: "2" kafka-pipeline: source: kafka: encryption: type: "ssl" topics: - name: "topic-name" group_id: "group-id" bootstrap_servers: - "bootstrap-server.us-east-1.aws.private.confluent.cloud:9092" authentication: sasl: plain: username: ${aws_secrets:confluent-kafka-secret:username} password: ${aws_secrets:confluent-kafka-secret:password} schema: type: confluent registry_url: https://my-registry.us-east-1.aws.confluent.cloud api_key: "${{aws_secrets:schema-secret:schema_registry_api_key}}" api_secret: "${{aws_secrets:schema-secret:schema_registry_api_secret}}" basic_auth_credentials_source: "USER_INFO" sink: - opensearch: hosts: ["https://search-mydomain.us-east-1.es.amazonaws.com"] aws: region: "us-east-1" index: "confluent-index" extension: aws: secrets: confluent-kafka-secret: secret_id: "my-kafka-secret" region: "us-east-1" schema-secret: secret_id: "my-self-managed-kafka-schema" region: "us-east-2" ``` # 将 OpenSearch 摄取管道与 Amazon Managed Streaming for Apache Kafka 你可以使用 [Kafka 插件](https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sources/kafka/)将来自亚马逊 Apache Managed [Streaming for Apache Kafka（亚马逊 MS OpenSearch K）的数据提取到你的摄取管](https://docs.aws.amazon.com/msk/latest/developerguide/)道中。在 Amazon MSK 中，您可以构建并运行使用 Apache Kafka 的应用程序来处理流数据。 OpenSearch Ingestion AWS PrivateLink 用于连接亚马逊 MSK。您可以从 Amazon MSK 和 Amazon MSK 无服务器集群摄取数据。这两个流程之间的唯一区别是在设置管道之前必须执行的先决条件步骤。 **Topics** + [预置的 Amazon MSK 先决条件](#msk-prereqs) + [Amazon MSK 无服务器先决条件](#msk-serverless-prereqs) + [步骤 1：配置管道角色](#msk-pipeline-role) + [步骤 2：创建管道](#msk-pipeline) + [步骤 3：（可选）使用 AWS Glue 架构注册表](#msk-glue) + [步骤 4：（可选）为 Amazon MSK 管道配置推荐的计算单位 (OCUs)](#msk-ocu) ## 预置的 Amazon MSK 先决条件 在创建 OpenSearch 摄取管道之前，请执行以下步骤： 1. 按照《Amazon Managed Streaming for Apache Kafka 开发人员指南》中 [Creating a cluster](https://docs.aws.amazon.com/msk/latest/developerguide/msk-create-cluster.html#create-cluster-console)** 说明的步骤，创建一个由 Amazon MSK 预置的集群。对于 **Broker 类型**，请选择除`t3`类型之外的任何选项，因为 OpenSearch Ingestion 不支持这些类型。 1. 集群处于**活动**状态后，请按照[开启多 VPC 连接](https://docs.aws.amazon.com/msk/latest/developerguide/aws-access-mult-vpc.html#mvpc-cluster-owner-action-turn-on)中的步骤执行操作。 1. 按照[将集群策略附加到 MSK 集群](https://docs.aws.amazon.com/msk/latest/developerguide/aws-access-mult-vpc.html#mvpc-cluster-owner-action-policy)的步骤附加以下策略之一，具体取决于集群与管道是否位于同一 AWS 账户。此策略允许 OpenSearch Ingestion 创建与您的 Amazon MSK 集群的 AWS PrivateLink 连接并从 Kafka 主题中读取数据。确保使用自身 ARN 更新 `resource`。 当集群与管道位于同一 AWS 账户时，适用以下策略： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "osis.amazonaws.com" }, "Action": [ "kafka:CreateVpcConnection", "kafka:DescribeClusterV2" ], "Resource": "arn:aws:kafka:us-east-1:111122223333:cluster/cluster-name/cluster-id" }, { "Effect": "Allow", "Principal": { "Service": "osis-pipelines.amazonaws.com" }, "Action": [ "kafka:CreateVpcConnection", "kafka:GetBootstrapBrokers", "kafka:DescribeClusterV2" ], "Resource": "arn:aws:kafka:us-east-1:111122223333:cluster/cluster-name/cluster-id" } ] } ``` ------ 如果您的 Amazon MSK 集群与您的管道 AWS 账户 不同，请改为附加以下策略。请注意，只有预置的 Amazon MSK 集群才能进行跨账户访问，Amazon MSK 无服务器集群不支持跨账户访问。的 ARN AWS `principal` 应该是您为工作流配置提供的相同管道角色的 ARN： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "osis.amazonaws.com" }, "Action": [ "kafka:CreateVpcConnection", "kafka:DescribeClusterV2" ], "Resource": "arn:aws:kafka:us-east-1:111122223333:cluster/cluster-name/cluster-id" }, { "Effect": "Allow", "Principal": { "Service": "osis-pipelines.amazonaws.com" }, "Action": [ "kafka:CreateVpcConnection", "kafka:GetBootstrapBrokers", "kafka:DescribeClusterV2" ], "Resource": "arn:aws:kafka:us-east-1:111122223333:cluster/cluster-name/cluster-id" }, { "Effect": "Allow", "Principal": { "AWS": "arn:aws:iam::444455556666:role/pipeline-role" }, "Action": [ "kafka-cluster:*", "kafka:*" ], "Resource": [ "arn:aws:kafka:us-east-1:111122223333:cluster/cluster-name/cluster-id", "arn:aws:kafka:us-east-1:111122223333:topic/cluster-name/cluster-id/*", "arn:aws:kafka:us-east-1:111122223333:group/cluster-name/*" ] } ] } ``` ------ 1. 按照[创建主题](https://docs.aws.amazon.com/msk/latest/developerguide/create-topic.html)中的步骤创建 Kafka 主题。确保`BootstrapServerString`这是私有终端节点（单 VPC）引导程序 URLs之一。`--replication-factor` 的值应为 `2` 或 `3`，具体取决于 Amazon MSK 集群包含的可用区数量。`--partitions` 的值至少应为 `10`。 1. 按照[生成和使用数据](https://docs.aws.amazon.com/msk/latest/developerguide/produce-consume.html)中的步骤生成和使用数据。再说一遍，请确保`BootstrapServerString`这是您的私有终端节点（单 VPC）引导程序 URLs之一。 ## Amazon MSK 无服务器先决条件 在创建 OpenSearch 摄取管道之前，请执行以下步骤： 1. 按照《Amazon Managed Streaming for Apache Kafka 开发人员指南》中 [Create an MSK Serverless cluster](https://docs.aws.amazon.com/msk/latest/developerguide/create-serverless-cluster.html#)** 说明的步骤，创建一个 Amazon MSK 无服务器集群。 1. 集群处于**活动**状态后，按照 [Attach a cluster policy to the MSK cluster](https://docs.aws.amazon.com/msk/latest/developerguide/aws-access-mult-vpc.html#mvpc-cluster-owner-action-policy) 中的步骤附加以下策略。确保使用自身 ARN 更新 `resource`。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "osis.amazonaws.com" }, "Action": [ "kafka:CreateVpcConnection", "kafka:DescribeClusterV2" ], "Resource": "arn:aws:kafka:us-east-1:111122223333:cluster/cluster-name/cluster-id" }, { "Effect": "Allow", "Principal": { "Service": "osis-pipelines.amazonaws.com" }, "Action": [ "kafka:CreateVpcConnection", "kafka:GetBootstrapBrokers", "kafka:DescribeClusterV2" ], "Resource": "arn:aws:kafka:us-east-1:111122223333:cluster/cluster-name/cluster-id" } ] } ``` ------ 此策略允许 OpenSearch Ingestion 创建与您的 Amazon MSK 无服务器集群的 AWS PrivateLink 连接并从 Kafka 主题中读取数据。当您的集群和管道处于相同状态时，此政策适用 AWS 账户，这必须是正确的，因为 Amazon MSK Serverless 不支持跨账户访问。 1. 按照[创建主题](https://docs.aws.amazon.com/msk/latest/developerguide/msk-serverless-create-topic.html)中的步骤创建 Kafka 主题。确保`BootstrapServerString`这是您的简单身份验证和安全层 (SASL) IAM 引导程序 URLs之一。`--replication-factor` 的值应为 `2` 或 `3`，具体取决于 Amazon MSK 无服务器集群包含的可用区数量。`--partitions` 的值至少应为 `10`。 1. 按照[生成和使用数据](https://docs.aws.amazon.com/msk/latest/developerguide/msk-serverless-produce-consume.html)中的步骤生成和使用数据。再说一遍，请确保`BootstrapServerString`这是您的简单身份验证和安全层 (SASL) IAM 引导程序 URLs之一。 ## 步骤 1：配置管道角色 预置好 Amazon MSK 集群或设置好无服务器集群后，在管道角色中添加要在管道配置中使用的以下 Kafka 权限： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "kafka-cluster:Connect", "kafka-cluster:AlterCluster", "kafka-cluster:DescribeCluster", "kafka:DescribeClusterV2", "kafka:GetBootstrapBrokers" ], "Resource": [ "arn:aws:kafka:us-east-1:111122223333:cluster/cluster-name/cluster-id" ] }, { "Effect": "Allow", "Action": [ "kafka-cluster:*Topic*", "kafka-cluster:ReadData" ], "Resource": [ "arn:aws:kafka:us-east-1:111122223333:topic/cluster-name/cluster-id/topic-name" ] }, { "Effect": "Allow", "Action": [ "kafka-cluster:AlterGroup", "kafka-cluster:DescribeGroup" ], "Resource": [ "arn:aws:kafka:us-east-1:111122223333:group/cluster-name/*" ] } ] } ``` ------ ## 步骤 2：创建管道 然后，你可以配置如下所示的 OpenSearch 摄取管道，将 Kafka 指定为来源： ``` version: "2" log-pipeline: source: kafka: acknowledgements: true topics: - name: "topic-name" group_id: "grouplambd-id" aws: msk: arn: "arn:aws:kafka:region:account-id:cluster/cluster-name/cluster-id" region: "us-west-2" processor: - grok: match: message: - "%{COMMONAPACHELOG}" - date: destination: "@timestamp" from_time_received: true sink: - opensearch: hosts: ["https://search-domain-endpoint.us-east-1es.amazonaws.com"] index: "index_name" aws_region: "region" aws_sigv4: true ``` 您可以使用预先配置的 Amazon MSK 蓝图，以创建此管道。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 ## 步骤 3：（可选）使用 AWS Glue 架构注册表 当您将 OpenSearch Ingestion 与 Amazon MSK 配合使用时，可以将 AVRO 数据格式用于架构注册表中托管的架构。 AWS Glue 在 [AWS Glue 架构注册表](https://docs.aws.amazon.com/glue/latest/dg/schema-registry.html)中，您可以集中发现、控制和演变数据流架构。 要使用此选项，请在管道配置中启用架构 `type`： ``` schema: type: "aws_glue" ``` 您还必须在您的管道角色中提供 AWS Glue 读取访问权限。您可以使用名为的 AWS 托管策略[AWSGlueSchemaRegistryReadonlyAccess](https://docs.aws.amazon.com/aws-managed-policy/latest/reference/AWSGlueSchemaRegistryReadonlyAccess.html)。此外，您的注册表必须 AWS 账户 与您的 OpenSearch 摄取管道位于同一区域中。 ## 步骤 4：（可选）为 Amazon MSK 管道配置推荐的计算单位 (OCUs) 每个计算单位的每个主题有一个使用者。代理在给定主题的使用者之间均衡分配分区。但是，当分区数量大于使用者数量时，Amazon MSK 将要求每个使用者托管多个分区。 OpenSearch Ingestion 具有内置的 auto Scaling，可以根据 CPU 使用率或管道中的待处理记录数量向上或向下扩展。 为实现最佳性能，请将分区分布在多个计算单位中以便并行处理。如果主题有大量分区（例如，超过 96 个，这是 OCUs 每个管道的最大分区），我们建议您将管道配置为 1— OCUs 96。因为它将根据需要自动扩缩。如果主题包含的分区数量较少（例如，少于 96 个），则最大计算单位应与分区数量相同。 当管道包含多个主题时，请选择分区数最多的主题作为参考来配置最大计算单位。通过向同一主题和使用者组添加另一个具有新集的 OCUs 管道，您可以几乎线性地扩展吞吐量。 # 在 Amazon RDS OpenSearch 中使用采集管道 您可以将 OpenSearch 采集管道与 Amazon RDS 配合使用，将现有数据和流更改（例如创建、更新和删除）导出到亚马逊 OpenSearch 服务域和集合。 OpenSearch Ingestion 管道整合了变更数据捕获 (CDC) 基础设施，以提供一种高规模、低延迟的方式来持续流式传输来自 Amazon RDS 的数据。支持 RDS for MySQL 和 RDS for PostgreSQL 您可以通过两种方式将 Amazon RDS 作为处理数据的来源：有或没有完整初始快照。完整初始快照是指定表的快照，此快照将导出至 Amazon S3。从那里，In OpenSearch gestion 管道将其发送到域中的一个索引，或者将其分区到域中的多个索引。为了保持 Amazon RDS 中的数据并保持 OpenSearch 一致，该管道将 Amazon RDS 实例中表中的所有创建、更新和删除事件与保存在一个或多个索引中的 OpenSearch 文档同步。 当您使用完整的初始快照时，您的 OpenSearch 摄取管道会首先提取快照，然后开始从 Amazon RDS 更改流中读取数据。它最终会赶上并保持 Amazon RDS 和 OpenSearch之间近乎实时的数据一致性。 您还可以使用 OpenSearch Ingestion 与 Amazon RDS 的集成来跟踪变更数据捕获并将所有更新收录到 Aurora 中。 OpenSearch如果您已通过其他机制获取完整快照，或仅需捕获 Amazon RDS 实例中数据的所有更改，则选择此选项。 选择此选项时，您需要[配置 Amazon RDS for MySQL 二进制日志记录](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_LogAccess.MySQL.BinaryFormat.html)或[为 Amazon RDS for PostgresSQL 数据库实例设置逻辑复制](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Appendix.PostgreSQL.CommonDBATasks.pglogical.setup-replication.html)。 **Topics** + [RDS for MySQL](rds-mysql.md) + [RDS for PostgreSQL](rds-PostgreSQL.md) # RDS for MySQL 完成以下步骤，使用适用于 MySQL 的 Amazon RDS for RDS 配置 OpenSearch 摄取管道。 **Topics** + [RDS for MySQL 先决条件](#rds-mysql-prereqs) + [步骤 1：配置管道角色](#rds-mysql-pipeline-role) + [步骤 2：创建管道](#rds-mysql-pipeline) + [数据一致性](#rds-mysql-pipeline-consistency) + [映射数据类型](#rds-mysql-pipeline-mapping) + [限制](#rds-mysql-pipeline-limitations) + [推荐 CloudWatch 警报](#aurora-mysql-pipeline-metrics) ## RDS for MySQL 先决条件 在创建 OpenSearch 摄取管道之前，请执行以下步骤： 1. 在 Amazon RDS 中创建自定义数据库参数组，以配置二进制日志记录并设置以下参数。 ``` binlog_format=ROW binlog_row_image=full binlog_row_metadata=FULL ``` 此外，确保 `binlog_row_value_options` 参数未设置为 `PARTIAL_JSON`。 有关更多信息，请参阅[配置 RDS for MySQL 二进制日志记录](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_LogAccess.MySQL.BinaryFormat.html)。 1. [选择或创建 RDS for MySQL 数据库实例](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_CreateDBInstance.html)，并将上一步创建的参数组与该数据库实例相关联。 1. 验证数据库是否已启用自动备份。有关更多信息，请参阅[启用自动备份](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_WorkingWithAutomatedBackups.Enabling.html)。 1. 配置二进制日志保留时间，确保留出足够的时间进行复制，例如 24 小时。有关更多信息，请参阅《Amazon RDS 用户指南》**中的 [Setting and showing binary log configuration](https://docs.aws.amazon.com//AmazonRDS/latest/UserGuide/mysql-stored-proc-configuring.html)。 1. 使用 [Amazon RDS 和 AWS Secrets Manager的密码管理](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/rds-secrets-manager.html)，在 Amazon RDS 实例上设置用户名和密码身份验证。您也可以通过创建 Secrets [Manager 密钥来创建 username/password ](https://docs.aws.amazon.com/secretsmanager/latest/userguide/create_secret.html)组合。 1. 如果您使用完整的初始快照功能，请创建一个 AWS KMS key 和一个 IAM 角色来将数据从 Amazon RDS 导出到 Amazon S3。 IAM 角色还必须具有以下权限策略： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "ExportPolicy", "Effect": "Allow", "Action": [ "s3:PutObject*", "s3:ListBucket", "s3:GetObject*", "s3:DeleteObject*", "s3:GetBucketLocation" ], "Resource": [ "arn:aws:s3:::s3-bucket-used-in-pipeline", "arn:aws:s3:::s3-bucket-used-in-pipeline/*" ] } ] } ``` ------ 该角色还应拥有以下信任关系： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "export.rds.amazonaws.com" }, "Action": "sts:AssumeRole" } ] } ``` ------ 1. 选择或创建 OpenSearch 服务域或 OpenSearch 无服务器集合。有关更多信息，请参阅[创建 OpenSearch 服务域](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/createupdatedomains.html#createdomains)和[创建集合](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/serverless-manage.html#serverless-create)。 1. 将[基于资源的策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/ac.html#ac-types-resource)附加到域，或将[数据访问策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/serverless-data-access.html)附加到集合。这些访问策略允许 OpenSearch Ingestion 将数据从您的 Amazon RDS 数据库实例写入您的域或集合。 ## 步骤 1：配置管道角色 完成 Amazon RDS 管道先决条件设置后，[配置管道角色](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/pipeline-security-overview.html#pipeline-security-sink)，以便在管道配置中使用。还需为该角色添加以下针对 Amazon RDS 源的权限： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "allowReadingFromS3Buckets", "Effect": "Allow", "Action": [ "s3:GetObject", "s3:DeleteObject", "s3:GetBucketLocation", "s3:ListBucket", "s3:PutObject" ], "Resource": [ "arn:aws:s3:::s3_bucket", "arn:aws:s3:::s3_bucket/*" ] }, { "Sid": "allowNetworkInterfacesActions", "Effect": "Allow", "Action": [ "ec2:AttachNetworkInterface", "ec2:CreateNetworkInterface", "ec2:CreateNetworkInterfacePermission", "ec2:DeleteNetworkInterface", "ec2:DeleteNetworkInterfacePermission", "ec2:DetachNetworkInterface", "ec2:DescribeNetworkInterfaces" ], "Resource": [ "arn:aws:ec2:*:111122223333:network-interface/*", "arn:aws:ec2:*:111122223333:subnet/*", "arn:aws:ec2:*:111122223333:security-group/*" ] }, { "Sid": "allowDescribeEC2", "Effect": "Allow", "Action": [ "ec2:Describe*" ], "Resource": "*" }, { "Sid": "allowTagCreation", "Effect": "Allow", "Action": [ "ec2:CreateTags" ], "Resource": "arn:aws:ec2:*:111122223333:network-interface/*", "Condition": { "StringEquals": { "aws:RequestTag/OSISManaged": "true" } } }, { "Sid": "AllowDescribeInstances", "Effect": "Allow", "Action": [ "rds:DescribeDBInstances" ], "Resource": [ "arn:aws:rds:us-east-2:111122223333:db:*" ] }, { "Sid": "AllowSnapshots", "Effect": "Allow", "Action": [ "rds:DescribeDBSnapshots", "rds:CreateDBSnapshot", "rds:AddTagsToResource" ], "Resource": [ "arn:aws:rds:us-east-2:111122223333:db:DB-id", "arn:aws:rds:us-east-2:111122223333:snapshot:DB-id*" ] }, { "Sid": "AllowExport", "Effect": "Allow", "Action": [ "rds:StartExportTask" ], "Resource": [ "arn:aws:rds:us-east-2:111122223333:snapshot:DB-id*" ] }, { "Sid": "AllowDescribeExports", "Effect": "Allow", "Action": [ "rds:DescribeExportTasks" ], "Resource": "*", "Condition": { "StringEquals": { "aws:RequestedRegion": "us-east-2", "aws:ResourceAccount": "111122223333" } } }, { "Sid": "AllowAccessToKmsForExport", "Effect": "Allow", "Action": [ "kms:Decrypt", "kms:Encrypt", "kms:DescribeKey", "kms:RetireGrant", "kms:CreateGrant", "kms:ReEncrypt*", "kms:GenerateDataKey*" ], "Resource": [ "arn:aws:kms:us-east-2:111122223333:key/export-key-id" ] }, { "Sid": "AllowPassingExportRole", "Effect": "Allow", "Action": "iam:PassRole", "Resource": [ "arn:aws:iam::111122223333:role/export-role" ] }, { "Sid": "SecretsManagerReadAccess", "Effect": "Allow", "Action": [ "secretsmanager:GetSecretValue" ], "Resource": [ "arn:aws:secretsmanager:*:111122223333:secret:*" ] } ] } ``` ------ ## 步骤 2：创建管道 配置类似于以下 OpenSearch 内容的摄取管道。示例管道将 Amazon RDS 实例指定为源。 ``` version: "2" rds-mysql-pipeline: source: rds: db_identifier: "instance-id" engine: mysql database: "database-name" tables: include: - "table1" - "table2" s3_bucket: "bucket-name" s3_region: "bucket-region" s3_prefix: "prefix-name" export: kms_key_id: "kms-key-id" iam_role_arn: "export-role-arn" stream: true aws: sts_role_arn: "arn:aws:iam::account-id:role/pipeline-role" region: "us-east-1" authentication: username: ${{aws_secrets:secret:username}} password: ${{aws_secrets:secret:password}} sink: - opensearch: hosts: ["https://search-mydomain.us-east-1.es.amazonaws.com"] index: "${getMetadata(\"table_name\")}" index_type: custom document_id: "${getMetadata(\"primary_key\")}" action: "${getMetadata(\"opensearch_action\")}" document_version: "${getMetadata(\"document_version\")}" document_version_type: "external" aws: sts_role_arn: "arn:aws:iam::account-id:role/pipeline-role" region: "us-east-1" extension: aws: secrets: secret: secret_id: "rds-secret-id" region: "us-east-1" sts_role_arn: "arn:aws:iam::account-id:role/pipeline-role" refresh_interval: PT1H ``` 您可以使用预先配置的 Amazon RDS 蓝图，以创建此管道。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 要使用 Amazon Aurora 作为源，您需要为管道配置 VPC 访问权限。您选择的 VPC 应与 Amazon Aurora 源使用的 VPC 相同。然后选择一个或多个子网以及一个或多个 VPC 安全组。请注意，该管道需要访问 Aurora MySQL 数据库的网络权限，因此您还需确认 Aurora 集群已配置 VPC 安全组，允许来自管道 VPC 安全组的入站流量访问数据库端口。有关更多信息，请参阅[使用安全组控制访问权限](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/Overview.RDSSecurityGroups.html)。 如果您使用创建管道，则还必须将管道连接到 VPC，才能使用 Amazon Aurora 作为来源。 AWS 管理控制台 要执行此操作，请找到**网络配置**部分，选择**连接到 VPC**，然后从提供的任意一个默认选项中选择 CIDR，或者选择自己的 CIDR。您可以按照 [RFC 1918 当前最佳实践](https://datatracker.ietf.org/doc/html/rfc1918)中的定义，使用私有地址空间中的任何 CIDR。 要提供自定义 CIDR，请从下拉菜单中选择**其他**。为避免 OpenSearch Ingestion 和 Amazon RDS 之间的 IP 地址冲突，请确保 Amazon RDS VPC CIDR 不同于用于摄取的 CIDR。 OpenSearch 有关更多信息，请参阅[为管道配置 VPC 访问权限](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/pipeline-security.html#pipeline-vpc-configure)。 ## 数据一致性 该管道通过持续轮询或接收来自 Amazon RDS 实例的更改以及更新 OpenSearch索引中的相应文档来确保数据一致性。 OpenSearch Ingestion 支持 end-to-end确认，以确保数据的持久性。管道读取快照或流时，它会动态创建分区以进行并行处理。当管道在摄取 OpenSearch 域或集合中的所有记录后收到确认信息时，该管道会将该分区标记为已完成。如果要收录到 OpenSearch 无服务器搜索集合中，可以在管道中生成文档 ID。如果要采集到 OpenSearch Serverless 时间序列集合，请注意管道不会生成文档 ID，因此您必须在工作流接收器配置`document_id: "${getMetadata(\"primary_key\")}"`中省略文档 ID。 In OpenSearch gestion 管道还将传入的事件操作映射到相应的批量索引操作中，以帮助采集文档。这样可以保持数据的一致性，因此 Amazon RDS 中的每一次数据更改都与中的 OpenSearch相应文档更改保持一致。 ## 映射数据类型 OpenSearch 摄取管道将 MySQL 数据类型映射到适合 OpenSearch 服务域或集合使用的表示形式。如果中未定义映射模板 OpenSearch，则会根据第一个发送的文档使用[动态映射 OpenSearch ](https://docs.opensearch.org/latest/field-types/#dynamic-mapping)自动确定字段类型。您还可以 OpenSearch 通过映射模板明确定义最适合您的字段类型。 下表列出了 MySQL 数据类型和相应的 OpenSearch 字段类型。 OpenSearch 如果未定义显式映射，则 “*默认 OpenSearch 字段类型*” 列将显示相应的字段类型。在这种情况下，使用动态映射 OpenSearch自动确定字段类型。“*推荐的 OpenSearch 字段类型*” 列是建议在映射模板中明确指定的相应字段类型。这些字段类型与 MySQL 中的数据类型更加一致，通常可以启用中提供的更好的搜索功能 OpenSearch。 | MySQL 数据类型 | 默认 OpenSearch 字段类型 | 推荐 OpenSearch 字段类型 | | --- | --- | --- | | BIGINT | 长整数 | 长整数 | | BIGINT UNSIGNED | 长整数 | 无符号长整数 | | BIT | 长整数 | 字节、短整数、整数或长整数，具体取决于位数 | | DECIMAL | 文本 | 双精度或关键字 | | DOUBLE | 浮点数 | double | | FLOAT | 浮点数 | 浮点数 | | INT | 长整数 | 整数 | | INT UNSIGNED | 长整数 | 长整数 | | MEDIUMINT | 长整数 | 整数 | | MEDIUMINT UNSIGNED | 长整数 | 整数 | | NUMERIC | 文本 | 双精度或关键字 | | SMALLINT | 长整数 | short | | SMALLINT UNSIGNED | 长整数 | 整数 | | TINYINT | 长整数 | 字节 | | TINYINT UNSIGNED | 长整数 | short | | BINARY | 文本 | binary | | BLOB | 文本 | binary | | CHAR | 文本 | 文本 | | ENUM | 文本 | 关键字 | | LONGBLOB | 文本 | binary | | LONGTEXT | 文本 | 文本 | | MEDIUMBLOB | 文本 | binary | | MEDIUMTEXT | 文本 | 文本 | | SET | 文本 | 关键字 | | TEXT | 文本 | 文本 | | TINYBLOB | 文本 | binary | | TINYTEXT | 文本 | 文本 | | VARBINARY | 文本 | binary | | VARCHAR | 文本 | 文本 | | DATE | 长整数（以 epoch 毫秒为单位） | date | | DATETIME | 长整数（以 epoch 毫秒为单位） | date | | TIME | 长整数（以 epoch 毫秒为单位） | date | | TIMESTAMP | 长整数（以 epoch 毫秒为单位） | date | | YEAR | 长整数（以 epoch 毫秒为单位） | date | | GEOMETRY | 文本（采用 WKT 格式） | geo\$1shape | | GEOMETRYCOLLECTION | 文本（采用 WKT 格式） | geo\$1shape | | LINESTRING | 文本（采用 WKT 格式） | geo\$1shape | | MULTILINESTRING | 文本（采用 WKT 格式） | geo\$1shape | | MULTIPOINT | 文本（采用 WKT 格式） | geo\$1shape | | MULTIPOLYGON | 文本（采用 WKT 格式） | geo\$1shape | | POINT | 文本（采用 WKT 格式） | geo\$1point 或 geo\$1shape | | POLYGON | 文本（采用 WKT 格式） | geo\$1shape | | JSON | 文本 | object | 我们建议您在摄取管道中配置死信队列 (DLQ)。 OpenSearch 如果您已配置队列，S OpenSearch ervice 会将所有因动态映射失败而无法载入的失败文档发送到队列。 如果自动映射失败，则可以在管道配置中使用 `template_type` 和 `template_content`，以定义显式映射规则。或者，您可以在启动管道之前直接在搜索域或集合中创建映射模板。 ## 限制 在为 RDS for MySQL 设置 OpenSearch 摄取管道时，请考虑以下限制： + 该集成仅支持每个管道一个 MySQL 数据库。 + 该集成目前不支持跨区域数据摄取；您的 Amazon RDS 实例和 OpenSearch 域必须相同。 AWS 区域 + 该集成目前不支持跨账户数据摄取；您的 Amazon RDS 实例和 OpenSearch 摄取管道必须处于相同状态。 AWS 账户 + 确保 Amazon RDS 实例已使用 Secrets Manager 启用身份验证，这是唯一受支持的身份验证机制。 + 无法更新现有的管道配置以从不同的数据库中提取不同 and/or 表的数据。要更新管道的数据库和/或表名称，则必须创建新的管道。 + 通常不支持数据定义语言（DDL）语句。以下情况下，数据一致性将无法得到保证： + 主键已更改 (add/delete/rename)。 + 表已被删除/截断。 + 列名或数据类型已发生更改。 + 如果要同步的 MySQL 表未定义主键，则无法保证数据一致性。您需要在 OpenSearch 接收器配置中正确定义自定义`document_id`选项才能同步 updates/deletes 到 OpenSearch。 + 不支持带有级联删除操作的外键引用，这可能会导致 RDS for MySQL 和之间的数据不一致。 OpenSearch + 不支持 Amazon RDS 多可用区数据库集群。 + 支持的版本：MySQL 版本 8.0 及更高版本。 ## 推荐 CloudWatch 警报 建议使用以下 CloudWatch 指标来监控您的摄取管道的性能。这些指标可帮助您确定处理的导出数据量、处理的流事件量、处理导出和流事件时的错误数以及写入目标的文档数量。您可以设置 CloudWatch 警报，以便在其中一个指标在指定时间内超过指定值时执行操作。 | 指标 | 说明 | | --- | --- | | pipeline-name.rds.凭证已更改 | 该指标表示 AWS 密钥的轮换频率。 | | pipeline-name.rds。 executorRefreshErrors | 此指标表示无法刷新 AWS 密钥。 | | pipeline-name.rds。 exportRecordsTotal | 该指标指示从 Amazon Aurora 导出的记录数量。 | | pipeline-name.rds。 exportRecordsProcessed | 该指标表示 OpenSearch 摄取管道处理的记录数。 | | pipeline-name.rds。 exportRecordProcessing错误 | 该指标表示从 Amazon Aurora 集群读取数据时 OpenSearch 摄取管道中出现的处理错误数量。 | | pipeline-name.rds。 exportRecordsSuccess总计 | 该指标指示成功处理的导出记录总数。 | | pipeline-name.rds。 exportRecordsFailed总计 | 该指标指示处理失败的导出记录总数。 | | pipeline-name已收到.rds.bytes | 该指标表示 OpenSearch 摄取管道接收的总字节数。 | | pipeline-name.rds.bytes已处理 | 该指标表示 OpenSearch 摄取管道处理的总字节数。 | | pipeline-name.rds。 streamRecordsSuccess总计 | 该指标指示该流中成功处理的记录总数。 | | pipeline-name.rds。 streamRecordsFailed总计 | 该指标指示该流中处理失败的记录总数。 | # RDS for PostgreSQL 完成以下步骤，使用适用于 Post OpenSearch greSQL 的 Amazon RDS for RDS 配置摄取管道。 **Topics** + [RDS for PostgreSQL 先决条件](#rds-PostgreSQL-prereqs) + [步骤 1：配置管道角色](#rds-mysql-pipeline-role) + [步骤 2：创建管道](#rds-PostgreSQL-pipeline) + [数据一致性](#rds-mysql-pipeline-consistency) + [映射数据类型](#rds-PostgreSQL-pipeline-mapping) + [限制](#rds-PostgreSQL-pipeline-limitations) + [推荐 CloudWatch 警报](#aurora-mysql-pipeline-metrics) ## RDS for PostgreSQL 先决条件 在创建 OpenSearch 摄取管道之前，请执行以下步骤： 1. 在 Amazon RDS 中[创建自定义数据库参数组](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/zero-etl.setting-up.html#zero-etl.parameters)以配置逻辑复制。 ``` rds.logical_replication=1 ``` 有关更多信息，请参阅 [Performing logical replication for Amazon RDS for PostgreSQL](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/PostgreSQL.Concepts.General.FeatureSupport.LogicalReplication.html)。 1. [选择或创建 RDS for PostgreSQL 数据库实例](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/CHAP_GettingStarted.CreatingConnecting.PostgreSQL.html)，并将步骤 1 创建的参数组与该数据库实例相关联。 1. 使用 [Amazon 和 AWS Secrets Manager的密码管理](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/rds-secrets-manager.html)，在 Amazon RDS 实例上设置用户名和密码身份验证。您也可以通过创建 Secrets [Manager 密钥来创建 username/password ](https://docs.aws.amazon.com/secretsmanager/latest/userguide/create_secret.html)组合。 1. 如果您使用完整的初始快照功能，请创建一个 AWS KMS key 和一个 IAM 角色来将数据从 Amazon RDS 导出到 Amazon S3。 IAM 角色还必须具有以下权限策略： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "ExportPolicy", "Effect": "Allow", "Action": [ "s3:PutObject*", "s3:ListBucket", "s3:GetObject*", "s3:DeleteObject*", "s3:GetBucketLocation" ], "Resource": [ "arn:aws:s3:::s3-bucket-used-in-pipeline", "arn:aws:s3:::s3-bucket-used-in-pipeline/*" ] } ] } ``` ------ 该角色还应拥有以下信任关系： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "export.rds.amazonaws.com" }, "Action": "sts:AssumeRole" } ] } ``` ------ 1. 选择或创建 OpenSearch 服务域或 OpenSearch 无服务器集合。有关更多信息，请参阅[创建 OpenSearch 服务域](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/createupdatedomains.html#createdomains)和[创建集合](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/serverless-manage.html#serverless-create)。 1. 将[基于资源的策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/ac.html#ac-types-resource)附加到域，或将[数据访问策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/serverless-data-access.html)附加到集合。这些访问策略允许 OpenSearch Ingestion 将数据从您的 Amazon RDS 数据库实例写入您的域或集合。 ## 步骤 1：配置管道角色 完成 Amazon RDS 管道先决条件设置后，[配置管道角色](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/pipeline-security-overview.html#pipeline-security-sink)，以便在管道配置中使用。还需为该角色添加以下针对 Amazon RDS 源的权限： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "allowReadingFromS3Buckets", "Effect": "Allow", "Action": [ "s3:GetObject", "s3:DeleteObject", "s3:GetBucketLocation", "s3:ListBucket", "s3:PutObject" ], "Resource": [ "arn:aws:s3:::s3_bucket", "arn:aws:s3:::s3_bucket/*" ] }, { "Sid": "allowNetworkInterfacesActions", "Effect": "Allow", "Action": [ "ec2:AttachNetworkInterface", "ec2:CreateNetworkInterface", "ec2:CreateNetworkInterfacePermission", "ec2:DeleteNetworkInterface", "ec2:DeleteNetworkInterfacePermission", "ec2:DetachNetworkInterface", "ec2:DescribeNetworkInterfaces" ], "Resource": [ "arn:aws:ec2:*:111122223333:network-interface/*", "arn:aws:ec2:*:111122223333:subnet/*", "arn:aws:ec2:*:111122223333:security-group/*" ] }, { "Sid": "allowDescribeEC2", "Effect": "Allow", "Action": [ "ec2:Describe*" ], "Resource": "*" }, { "Sid": "allowTagCreation", "Effect": "Allow", "Action": [ "ec2:CreateTags" ], "Resource": "arn:aws:ec2:*:111122223333:network-interface/*", "Condition": { "StringEquals": { "aws:RequestTag/OSISManaged": "true" } } }, { "Sid": "AllowDescribeInstances", "Effect": "Allow", "Action": [ "rds:DescribeDBInstances" ], "Resource": [ "arn:aws:rds:us-east-2:111122223333:db:*" ] }, { "Sid": "AllowSnapshots", "Effect": "Allow", "Action": [ "rds:DescribeDBSnapshots", "rds:CreateDBSnapshot", "rds:AddTagsToResource" ], "Resource": [ "arn:aws:rds:us-east-2:111122223333:db:DB-id", "arn:aws:rds:us-east-2:111122223333:snapshot:DB-id*" ] }, { "Sid": "AllowExport", "Effect": "Allow", "Action": [ "rds:StartExportTask" ], "Resource": [ "arn:aws:rds:us-east-2:111122223333:snapshot:DB-id*" ] }, { "Sid": "AllowDescribeExports", "Effect": "Allow", "Action": [ "rds:DescribeExportTasks" ], "Resource": "*", "Condition": { "StringEquals": { "aws:RequestedRegion": "us-east-2", "aws:ResourceAccount": "111122223333" } } }, { "Sid": "AllowAccessToKmsForExport", "Effect": "Allow", "Action": [ "kms:Decrypt", "kms:Encrypt", "kms:DescribeKey", "kms:RetireGrant", "kms:CreateGrant", "kms:ReEncrypt*", "kms:GenerateDataKey*" ], "Resource": [ "arn:aws:kms:us-east-2:111122223333:key/export-key-id" ] }, { "Sid": "AllowPassingExportRole", "Effect": "Allow", "Action": "iam:PassRole", "Resource": [ "arn:aws:iam::111122223333:role/export-role" ] }, { "Sid": "SecretsManagerReadAccess", "Effect": "Allow", "Action": [ "secretsmanager:GetSecretValue" ], "Resource": [ "arn:aws:secretsmanager:*:111122223333:secret:*" ] } ] } ``` ------ ## 步骤 2：创建管道 配置如下所示的 OpenSearch 摄取管道，该管道将适用于 PostgreSQL 的 RDS 实例指定为源。 ``` version: "2" rds-postgres-pipeline: source: rds: db_identifier: "instance-id" engine: postgresql database: "database-name" tables: include: - "schema1.table1" - "schema2.table2" s3_bucket: "bucket-name" s3_region: "bucket-region" s3_prefix: "prefix-name" export: kms_key_id: "kms-key-id" iam_role_arn: "export-role-arn" stream: true aws: sts_role_arn: "arn:aws:iam::account-id:role/pipeline-role" region: "us-east-1" authentication: username: ${{aws_secrets:secret:username}} password: ${{aws_secrets:secret:password}} sink: - opensearch: hosts: ["https://search-mydomain.us-east-1.es.amazonaws.com"] index: "${getMetadata(\"table_name\")}" index_type: custom document_id: "${getMetadata(\"primary_key\")}" action: "${getMetadata(\"opensearch_action\")}" document_version: "${getMetadata(\"document_version\")}" document_version_type: "external" aws: sts_role_arn: "arn:aws:iam::account-id:role/pipeline-role" region: "us-east-1" extension: aws: secrets: secret: secret_id: "rds-secret-id" region: "us-east-1" sts_role_arn: "arn:aws:iam::account-id:role/pipeline-role" refresh_interval: PT1H ``` **注意** 您可以使用预先配置的 Amazon RDS 蓝图，以创建此管道。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 要使用 Amazon Aurora 作为源，您需要为管道配置 VPC 访问权限。您选择的 VPC 应与 Amazon Aurora 源使用的 VPC 相同。然后选择一个或多个子网以及一个或多个 VPC 安全组。请注意，该管道需要访问 Aurora MySQL 数据库的网络权限，因此您还需确认 Aurora 集群已配置 VPC 安全组，允许来自管道 VPC 安全组的入站流量访问数据库端口。有关更多信息，请参阅[使用安全组控制访问权限](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/Overview.RDSSecurityGroups.html)。 如果您使用创建管道，则还必须将管道连接到 VPC，才能使用 Amazon Aurora 作为来源。 AWS 管理控制台 要执行此操作，请找到**网络配置**部分，选择**连接到 VPC**，然后从提供的任意一个默认选项中选择 CIDR，或者选择自己的 CIDR。您可以按照 [RFC 1918 当前最佳实践](https://datatracker.ietf.org/doc/html/rfc1918)中的定义，使用私有地址空间中的任何 CIDR。 要提供自定义 CIDR，请从下拉菜单中选择**其他**。为避免 OpenSearch Ingestion 和 Amazon RDS 之间的 IP 地址冲突，请确保 Amazon Aurora VPC CIDR 与用于摄取的 CIDR 不同。 OpenSearch 有关更多信息，请参阅[为管道配置 VPC 访问权限](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/pipeline-security.html#pipeline-vpc-configure)。 ## 数据一致性 该管道通过持续轮询或接收来自 Amazon RDS 实例的更改以及更新 OpenSearch索引中的相应文档来确保数据一致性。 OpenSearch Ingestion 支持 end-to-end确认，以确保数据的持久性。管道读取快照或流时，它会动态创建分区以进行并行处理。当管道在摄取 OpenSearch 域或集合中的所有记录后收到确认信息时，该管道会将该分区标记为已完成。如果要收录到 OpenSearch 无服务器搜索集合中，可以在管道中生成文档 ID。如果要采集到 OpenSearch Serverless 时间序列集合，请注意管道不会生成文档 ID，因此您必须在工作流接收器配置`document_id: "${getMetadata(\"primary_key\")}"`中省略文档 ID。 In OpenSearch gestion 管道还将传入的事件操作映射到相应的批量索引操作中，以帮助采集文档。这样可以保持数据的一致性，因此 Amazon RDS 中的每一次数据更改都与中的 OpenSearch相应文档更改保持一致。 ## 映射数据类型 OpenSearch 摄取管道将 PostgreSQL 数据类型映射到适合 OpenSearch 服务域或集合使用的表示形式。如果中未定义映射模板 OpenSearch，则根据第一个发送的文档使用[动态映射 OpenSearch ](https://docs.opensearch.org/latest/field-types/#dynamic-mapping)自动确定字段类型。您还可以 OpenSearch 通过映射模板明确定义最适合您的字段类型。 下表列出了适用于 PostgreSQL 的 RDS 数据类型和相应的字段类型。 OpenSearch OpenSearch 如果未定义显式映射，则 “*默认 OpenSearch 字段类型*” 列将显示相应的字段类型。在这种情况下，使用动态映射 OpenSearch 自动确定字段类型。“*推荐 OpenSearch 字段类型*” 列是在映射模板中明确指定的相应推荐字段类型。这些字段类型与 RDS for PostgreSQL 中的数据类型更加一致，通常可以启用中提供的更好的搜索功能。 OpenSearch | RDS for PostgreSQL 数据类型 | 默认 OpenSearch 字段类型 | 推荐 OpenSearch 字段类型 | | --- | --- | --- | | smallint | 长整数 | short | | 整数 | 长整数 | integer | | bigint | 长整数 | 长整数 | | decimal | 文本 | 双精度或关键字 | | numeric[ (p, s) ] | 文本 | 双精度或关键字 | | real | float | 浮点数 | | double precision | 浮点数 | double | | smallserial | 长整数 | short | | Serial | 长整数 | 整数 | | bigserial | 长整数 | 长整数 | | money | object | object | | character varying(n) | 文本 | 文本 | | varchar(n) | 文本 | 文本 | | character(n) | 文本 | 文本 | | char(n) | 文本 | 文本 | | bpchar(n) | 文本 | 文本 | | bpchar | 文本 | 文本 | | 文本 | 文本 | 文本 | | enum | 文本 | 文本 | | bytea | 文本 | binary | | timestamp [ (p) ] [没有时区] | 长整数（以 epoch 毫秒为单位） | date | | 有时区的 timestamp [ (p) ] | 长整数（以 epoch 毫秒为单位） | date | | date | 长整数（以 epoch 毫秒为单位） | date | | time [ (p) ] [没有时区] | 长整数（以 epoch 毫秒为单位） | date | | 有时区的 time [ (p) ] | 长整数（以 epoch 毫秒为单位） | date | | interval [ fields ] [ (p) ] | 文本（ISO8601 格式） | 文本 | | 布尔值 | 布尔值 | 布尔值 | | point | 文本（采用 WKT 格式） | geo\$1shape | | 折线图 | 文本（采用 WKT 格式） | geo\$1shape | | lseg | 文本（采用 WKT 格式） | geo\$1shape | | box | 文本（采用 WKT 格式） | geo\$1shape | | path | 文本（采用 WKT 格式） | geo\$1shape | | 多边形 | 文本（采用 WKT 格式） | geo\$1shape | | circle | object | object | | cidr | 文本 | 文本 | | inet | 文本 | 文本 | | macaddr | 文本 | 文本 | | macaddr8 | 文本 | 文本 | | bit(n) | 长整数 | 字节、短整数、整数或长整数（具体取决于位数） | | bit varying(n) | 长整数 | 字节、短整数、整数或长整数（具体取决于位数） | | json | object | object | | jsonb | object | object | | jsonpath | 文本 | 文本 | 我们建议您在摄取管道中配置死信队列 (DLQ)。 OpenSearch 如果您已配置队列，S OpenSearch ervice 会将所有因动态映射失败而无法载入的失败文档发送到队列。 如果自动映射失败，则可以在管道配置中使用 `template_type` 和 `template_content` 来定义显式映射规则。或者，您可以在启动管道之前直接在搜索域或集合中创建映射模板。 ## 限制 在为 RDS for PostgreSQL 设置 OpenSearch 摄取管道时，请考虑以下限制： + 该集成仅支持每个管道一个 PostgreSQL 数据库。 + 该集成目前不支持跨区域数据摄取；您的 Amazon RDS 实例和 OpenSearch 域必须相同。 AWS 区域 + 该集成目前不支持跨账户数据摄取；您的 Amazon RDS 实例和 OpenSearch 摄取管道必须处于相同状态。 AWS 账户 + 确保 Amazon RDS 实例启用了使用身份验证 AWS Secrets Manager，这是唯一支持的身份验证机制。 + 无法更新现有的管道配置以从不同的数据库中提取不同 and/or 表的数据。要更新管道的数据库和/或表名称，则必须停止该管道，然后使用更新后的配置重新启动，或者创建新管道。 + 通常不支持数据定义语言（DDL）语句。以下情况下，数据一致性将无法得到保证： + 主键已更改 (add/delete/rename)。 + 表已被删除/截断。 + 列名或数据类型已发生更改。 + 如果要同步的 PostgreSQL 表未定义主键，则无法保证数据一致性。你需要在中定义自定义`document_id`选项 OpenSearch 和接收器配置才能同步 updates/deletes 到 OpenSearch。 + 不支持 RDS 多可用区数据库集群。 + 支持的版本：PostgreSQL 16 及更高版本。 ## 推荐 CloudWatch 警报 建议使用以下 CloudWatch 指标来监控您的摄取管道的性能。这些指标可帮助您确定处理的导出数据量、处理的流事件量、处理导出和流事件时的错误数以及写入目标的文档数量。您可以设置 CloudWatch 警报，以便在其中一个指标在指定时间内超过指定值时执行操作。 | 指标 | 说明 | | --- | --- | | pipeline-name.rds.凭证已更改 | 该指标表示 AWS 密钥的轮换频率。 | | pipeline-name.rds。 executorRefreshErrors | 此指标表示无法刷新 AWS 密钥。 | | pipeline-name.rds。 exportRecordsTotal | 该指标指示从 Amazon Aurora 导出的记录数量。 | | pipeline-name.rds。 exportRecordsProcessed | 该指标表示 OpenSearch 摄取管道处理的记录数。 | | pipeline-name.rds。 exportRecordProcessing错误 | 该指标表示从 Amazon Aurora 集群读取数据时 OpenSearch 摄取管道中出现的处理错误数量。 | | pipeline-name.rds。 exportRecordsSuccess总计 | 该指标指示成功处理的导出记录总数。 | | pipeline-name.rds。 exportRecordsFailed总计 | 该指标指示处理失败的导出记录总数。 | | pipeline-name已收到.rds.bytes | 该指标表示 OpenSearch 摄取管道接收的总字节数。 | | pipeline-name.rds.bytes已处理 | 该指标表示 OpenSearch 摄取管道处理的总字节数。 | | pipeline-name.rds。 streamRecordsSuccess总计 | 该指标指示该流中成功处理的记录总数。 | | pipeline-name.rds。 streamRecordsFailed总计 | 该指标指示该流中处理失败的记录总数。 | # 在 Amazon S3 中使用 OpenSearch 摄取管道 借助 OpenSearch Ingestion，您可以将 Amazon S3 用作源或目标。当您使用 Amazon S3 作为数据源时，会将数据发送到 OpenSearch 摄取管道。当您使用 Amazon S3 作为目标时，会将数据从 OpenSearch 摄取管道写入到一个或多个 S3 存储桶。 **Topics** + [亚马逊 S3 作为来源](#s3-source) + [Amazon S3 作为目标](#s3-destination) + [将 Amazon S3 跨账户作为源](#fdsf) ## 亚马逊 S3 作为来源 您可以通过两种方式使用 Amazon S3 作为源处理数据：*S3-SQS 处理*和*计划扫描*。 如果您需要在文件写入 S3 后近实时扫描文件，请使用 S3-SQS 处理。您可以配置 Amazon S3 存储桶，在存储桶中存储或修改对象时随时触发事件。使用一次性扫描或定期计划扫描批处理 S3 存储桶中的数据。 **Topics** + [先决条件](#s3-prereqs) + [步骤 1：配置管道角色](#s3-pipeline-role) + [步骤 2：创建管道](#s3-pipeline) ### 先决条件 要使用 Amazon S3 作为预定扫描或 S3-SQS 处理的 OpenSearch 摄取管道的来源，[请先创建一个 S3 存储桶](https://docs.aws.amazon.com/AmazonS3/latest/userguide/create-bucket-overview.html)。 **注意** 如果 OpenSearch Ingestion 管道中用作源的 S3 存储桶位于不同的存储桶中 AWS 账户，则还需要对该存储桶启用跨账户读取权限。这样管道将可读取和处理数据。要启用跨账户权限，请参阅 *Amazon S3 用户指南*中的[存储桶拥有者授予跨账户存储桶权限](https://docs.aws.amazon.com/AmazonS3/latest/userguide/example-walkthroughs-managing-access-example2.html)。 如果您的 S3 存储桶位于多个账户中，请使用 `bucket_owners` 映射。有关示例，请参阅 OpenSearch文档中的[跨账户 S3 访问权限](https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sources/s3/#cross-account-s3-access)。 要设置 S3-SQS 处理，还需要执行以下步骤： 1. [创建 Amazon SQS 队列](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/step-create-queue.html)。 1. 在以 SQS 队列为目标的 S3 存储桶上[启用事件通知](https://docs.aws.amazon.com/AmazonS3/latest/userguide/enable-event-notifications.html)。 ### 步骤 1：配置管道角色 与其他将数据*推送*到管道的源插件不同，[S3 源插件](https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sources/s3/)采用基于读取的架构，管道从源中*拉取*数据。 因此，为使管道能够从 S3 读取，必须在管道的 S3 源配置中指定一个可以同时访问 S3 存储桶和 Amazon SQS 队列的角色。管道将担任此角色，以便从队列中读取数据。 **注意** 在 S3 源配置中指定的角色必须是[管道角色]()。因此，管道角色必须包含两个单独的权限策略，一个用于写入接收器，另一个用于从 S3 源中拉取。您必须在所有管道组件中使用相同的 `sts_role_arn`。 以下示例策略显示了使用 S3 作为源所需的权限： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action":[ "s3:ListBucket", "s3:GetBucketLocation", "s3:GetObject" ], "Resource": "arn:aws:s3:::amzn-s3-demo-bucket/*" }, { "Effect":"Allow", "Action":"s3:ListAllMyBuckets", "Resource":"arn:aws:s3:::*" }, { "Effect": "Allow", "Action": [ "sqs:DeleteMessage", "sqs:ReceiveMessage", "sqs:ChangeMessageVisibility" ], "Resource": "arn:aws:sqs:us-east-1:111122223333:MyS3EventSqsQueue" } ] } ``` ------ 必须将以下权限附加到在 S3 源插件配置的 `sts_role_arn` 选项中指定的 IAM 角色： ``` version: "2" source: s3: ... aws: ... processor: ... sink: - opensearch: ... ``` ### 步骤 2：创建管道 设置权限后，您可以根据您的 Amazon S3 用例配置 OpenSearch 摄取管道。 #### S3-SQS 处理 要设置 S3-SQS 处理，请配置您的管道，指定 S3 作为源并设置 Amazon SQS 通知： ``` version: "2" s3-pipeline: source: s3: notification_type: "sqs" codec: newline: null sqs: queue_url: "https://sqs.us-east-1amazonaws.com/account-id/ingestion-queue" compression: "none" aws: region: "region" processor: - grok: match: message: - "%{COMMONAPACHELOG}" - date: destination: "@timestamp" from_time_received: true sink: - opensearch: hosts: ["https://search-domain-endpoint.us-east-1es.amazonaws.com"] index: "index-name" aws: region: "region" ``` 如果您在处理 Amazon S3 上的小文件时发现 CPU 利用率较低，可考虑通过修改 `workers` 选项的值来提高吞吐量。有关更多信息，请参阅 [S3 plugin configuration options](https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sources/s3/#configuration)。 #### 计划扫描 要设置计划扫描，请使用适用于所有 S3 存储桶的扫描级别或存储桶级别的计划来配置管道。存储桶级别计划或扫描间隔配置始终覆盖扫描级别配置。 您可以使用*一次性扫描*（非常适合数据迁移）或*定期扫描*（非常适合批处理）配置计划扫描。 要将管道配置为从 Amazon S3 读取，请使用预先配置的 Amazon S3 蓝图。您可以编辑管道配置的 `scan` 部分以满足计划需求。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 **一次性扫描** 一次性计划扫描运行一次。在管道配置中，您可以使用 `start_time` 和 `end_time` 指定希望何时扫描存储桶中的对象。或者，您也可以使用 `range` 指定相对于当前时间的时间间隔，以该时间间隔扫描存储桶中的对象。 例如，范围设置为 `PT4H` 将扫描最近四个小时内创建的所有文件。要配置再次运行一次性扫描，必须先停止管道，然后再重新启动。如果未配置范围，则还必须更新开始时间和结束时间。 以下配置为所有存储桶及这些存储桶中的所有对象设置一次性扫描： ``` version: "2" log-pipeline: source: s3: codec: csv: compression: "none" aws: region: "region" acknowledgments: true scan: buckets: - bucket: name: my-bucket filter: include_prefix: - Objects1/ exclude_suffix: - .jpeg - .png - bucket: name: my-bucket-2 key_prefix: include: - Objects2/ exclude_suffix: - .jpeg - .png delete_s3_objects_on_read: false processor: - date: destination: "@timestamp" from_time_received: true sink: - opensearch: hosts: ["https://search-domain-endpoint.us-east-1es.amazonaws.com"] index: "index-name" aws: region: "region" dlq: s3: bucket: "dlq-bucket" region: "us-east-1" ``` 以下配置设置在指定时段内对所有存储桶运行一次性扫描。这意味着 S3 仅处理创建时间在此时段内的对象。 ``` scan: start_time: 2023-01-21T18:00:00.000Z end_time: 2023-04-21T18:00:00.000Z buckets: - bucket: name: my-bucket-1 filter: include: - Objects1/ exclude_suffix: - .jpeg - .png - bucket: name: my-bucket-2 filter: include: - Objects2/ exclude_suffix: - .jpeg - .png ``` 以下配置设置扫描级别和存储桶级别一次性扫描。存储桶级别开始时间和结束时间将覆盖扫描级别开始时间和结束时间。 ``` scan: start_time: 2023-01-21T18:00:00.000Z end_time: 2023-04-21T18:00:00.000Z buckets: - bucket: start_time: 2023-01-21T18:00:00.000Z end_time: 2023-04-21T18:00:00.000Z name: my-bucket-1 filter: include: - Objects1/ exclude_suffix: - .jpeg - .png - bucket: start_time: 2023-01-21T18:00:00.000Z end_time: 2023-04-21T18:00:00.000Z name: my-bucket-2 filter: include: - Objects2/ exclude_suffix: - .jpeg - .png ``` 停止管道会移除对管道在停止之前所扫描对象的任何先前引用。如果停止单个扫描管道，则将在启动后重新扫描所有对象，即使这些已被扫描。如果您需要停止单个扫描管道，建议在重新启动管道之前更改时间段。 如果您需要按开始时间和结束时间筛选对象，则停止和启动管道是唯一的选择。如果您不需要按开始时间和结束时间进行筛选，则可以按名称筛选对象。按名称筛选对象不需要停止和启动管道。要执行此操作，请使用 `include_prefix` 和 `exclude_suffix`。 **定期扫描** 定期计划扫描按定期计划时间间隔对您指定的 S3 存储桶运行扫描。只能在扫描级别配置间隔，因为不支持单独执行存储桶级别配置。 在管道配置中，`interval` 将指定定期扫描频率，范围介于 30 秒到 365 天之间。始终在创建管道时运行首次扫描。`count` 定义扫描实例总数。 以下配置设置定期扫描，两次扫描之间的延迟为 12 小时： ``` scan: scheduling: interval: PT12H count: 4 buckets: - bucket: name: my-bucket-1 filter: include: - Objects1/ exclude_suffix: - .jpeg - .png - bucket: name: my-bucket-2 filter: include: - Objects2/ exclude_suffix: - .jpeg - .png ``` ## Amazon S3 作为目标 [要将数据从 OpenSearch 摄取管道写入 S3 存储桶，请使用预配置的 S3 蓝图创建带有 S3 接收器的管道。](https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sinks/s3/)该管道将选择性数据路由到 OpenSearch 接收器，同时将所有数据发送到 S3 中进行存档。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 创建 S3 接收器时，您可以从各种不同的[接收器编解码器](https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sinks/s3/#codec)指定首选格式。例如，如果要以列式格式写入数据，请选择 Parquet 或 Avro 编解码器。如果您更喜欢基于行的格式，请选择 JSON 或 NDJSON。要将数据写入指定架构中的 S3，您还可以使用 [Avro](https://avro.apache.org/docs/current/specification/#schema-declaration) 格式在接收器编解码器中定义内联架构。 以下示例在 S3 接收器中定义内联架构： ``` - s3: codec: parquet: schema: > { "type" : "record", "namespace" : "org.vpcFlowLog.examples", "name" : "VpcFlowLog", "fields" : [ { "name" : "version", "type" : "string"}, { "name" : "srcport", "type": "int"}, { "name" : "dstport", "type": "int"}, { "name" : "start", "type": "int"}, { "name" : "end", "type": "int"}, { "name" : "protocol", "type": "int"}, { "name" : "packets", "type": "int"}, { "name" : "bytes", "type": "int"}, { "name" : "action", "type": "string"}, { "name" : "logStatus", "type" : "string"} ] } ``` 定义此架构时，请指定管道向接收器发送的不同类型事件中可能存在的所有键的超集。 例如，如果事件可能缺少键，则在架构中添加值为 `null` 的键。Null 值声明允许架构处理非统一数据（一些事件具有这些键，另一些事件则没有）。当传入事件确实存在这些键时，则将键值写入接收器。 此架构定义充当筛选器，仅允许将定义的键发送到接收器，并从传入事件中删除未定义的键。 您也可以在接收器中使用 `include_keys` 和 `exclude_keys` 筛选路由到其他接收器的数据。两个筛选器互斥，因此在架构中一次只能使用一个筛选器。此外，不能在用户定义的架构中使用它们。 要使用上述筛选条件创建管道，请使用预先配置的接收器筛选条件蓝图。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 ## 将 Amazon S3 跨账户作为源 您可以通过 Amazon S3 授予跨账户访问权限，这样 OpenSearch Ingestion 管道就可以访问另一个账户中的 S3 存储桶作为来源。要启用跨账户访问，请参阅《Amazon S3 用户指南》中的 [Bucket owner granting cross-account bucket permissions](https://docs.aws.amazon.com/AmazonS3/latest/userguide/example-walkthroughs-managing-access-example2.html)**。授予访问权限后，请确保您的管道角色具有所需的权限。 然后可以使用 `bucket_owners` 创建管道，从而对用作源的 Amazon S3 存储桶启用跨账户访问： ``` s3-pipeline: source: s3: notification_type: "sqs" codec: csv: delimiter: "," quote_character: "\"" detect_header: True sqs: queue_url: "https://sqs.ap-northeast-1.amazonaws.com/401447383613/test-s3-queue" bucket_owners: my-bucket-01: 123456789012 my-bucket-02: 999999999999 compression: "gzip" ``` # 在 Amaz OpenSearch on Security Lake 中使用采集管道 您可以使用 [S3 源插件](https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sources/s3/)将来自 A [mazon Security Lake](https://docs.aws.amazon.com/security-lake/latest/userguide/what-is-security-lake.html) 的数据提取到您的 OpenSearch 摄取管道中。Security Lake 会自动将来自 AWS 环境、本地环境和 SaaS 提供商的安全数据集中到专门构建的数据湖中。您可以创建订阅，将数据从 Security Lake 复制到您的 OpenSearch 摄取管道，然后由该渠道将其写入您的 OpenSearch 服务域或 OpenSearch 无服务器集合。 要将管道配置为从 Security Lake 读取，请使用预先配置的 Security Lake 蓝图。蓝图包括用于从 Security Lake 摄取开放式网络安全架构框架 (OCSF) parquet 文件的默认配置。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 **Topics** + [使用以 Amaz OpenSearch on Security Lake 为来源的摄取管道](configure-client-source-security-lake.md) + [使用以 Amaz OpenSearch on Security Lake 为接收器的摄取管道](configure-client-sink-security-lake.md) # 使用以 Amaz OpenSearch on Security Lake 为来源的摄取管道 您可以使用 OpenSearch Ingestion 管道中的 Amazon S3 源插件从 Amazon Security Lake 提取数据。Security Lake 自动将来自 AWS 环境、本地系统和 SaaS 提供商的安全数据集中到专用数据湖。 Amazon Security Lake 在管道内具有以下元数据属性： + `bucket_name`：Security Lake 创建的 Amazon S3 存储桶的名称，用于存储安全数据。 + `path_prefix`：在 Security Lake IAM 角色策略中定义的自定义源名称。 + `region`：Secur AWS 区域 ity Lake S3 存储桶所在的位置。 + `accountID`：启用安全湖的 AWS 账户 ID。 + `sts_role_arn`：用于与 Security Lake 结合使用的 IAM 角色的 ARN。 ## 先决条件 在创建 OpenSearch 摄取管道之前，请执行以下步骤： + [启用 Security Lake](https://docs.aws.amazon.com/security-lake/latest/userguide/getting-started.html#enable-service)。 + 在 Security Lake 中[创建订阅用户](https://docs.aws.amazon.com/security-lake/latest/userguide/subscriber-data-access.html#create-subscriber-data-access)。 + 选择要摄取到管道的源。 + 对于**订阅者凭证**，请添加您打算在 AWS 账户 哪里创建管道的 ID。对于外部 ID，请指定 `OpenSearchIngestion-{accountid}`。 + 对于**数据访问方法**，请选择 **S3**。 + 有关**通知详细信息**，请选择 **SQS 队列**。 创建订阅用户时，Security Lake 将自动创建两个内联权限策略，一个用于 S3，一个用于 SQS。策略采用以下格式：`AmazonSecurityLake-amzn-s3-demo-bucket-S3` 和 `AmazonSecurityLake-AWS Demo-SQS`。要允许管道访问订阅用户源，必须将必需权限与管道角色进行关联。 ## 配置管道角色 在 IAM 中创建新的权限策略，仅组合 Security Lake 自动创建的两个策略中的必需权限。以下示例策略显示了 OpenSearch 摄取管道从多个 Security Lake 来源读取数据所需的最低权限： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement":[ { "Effect":"Allow", "Action":[ "s3:GetObject" ], "Resource":[ "arn:aws:s3:::aws-security-data-lake-us-east-1-abcde/aws/LAMBDA_EXECUTION/1.0/*", "arn:aws:s3:::aws-security-data-lake-us-east-1-abcde/aws/S3_DATA/1.0/*", "arn:aws:s3:::aws-security-data-lake-us-east-1-abcde/aws/VPC_FLOW/1.0/*", "arn:aws:s3:::aws-security-data-lake-us-east-1-abcde/aws/ROUTE53/1.0/*", "arn:aws:s3:::aws-security-data-lake-us-east-1-abcde/aws/SH_FINDINGS/1.0/*" ] }, { "Effect":"Allow", "Action":[ "sqs:ReceiveMessage", "sqs:DeleteMessage" ], "Resource":[ "arn:aws:sqs:us-east-1:111122223333:AmazonSecurityLake-abcde-Main-Queue" ] } ] } ``` ------ **重要** Security Lake 不负责为您管理管道角色策略。如果向 Security Lake 订阅中添加源或从中删除源，则必须手动更新政策。Security Lake 将为每个日志源创建分区，因此您需要在管道角色中手动添加或删除权限。 必须将以下权限附加到在 S3 源插件配置的 `sts_role_arn` 选项的 `sqs` 下指定的 IAM 角色。 ``` version: "2" source: s3: ... sqs: queue_url: "https://sqs.us-east-1amazonaws.com/account-id/AmazonSecurityLake-amzn-s3-demo-bucket-Main-Queue" aws: ... processor: ... sink: - opensearch: ... ``` ## 创建管道 向管道角色添加权限后，使用预先配置的 Security Lake 蓝图创建管道。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 必须在 `s3` 源配置中指定 `queue_url` 选项，即要读取的 Amazon SQS 队列 URL。要设置 URL 格式，请在订阅用户配置中找到**订阅端点**，将 `arn:aws:` 更改为 `https://`。例如 `https://sqs.us-east-1amazonaws.com/account-id/AmazonSecurityLake-AWS Demo-Main-Queue`。 在 S3 源配置中指定的 `sts_role_arn` 必须是管道角色 ARN。 # 使用以 Amaz OpenSearch on Security Lake 为接收器的摄取管道 使用 OpenSearch Ingestion 中的 Amazon S3 接收器插件将数据从任何支持的来源发送到亚马逊安全湖。Security Lake 在专用数据湖中收集和存储来自 AWS本地环境和 SaaS 提供商的安全数据。 要配置管道以将日志数据写入 Security Lake，请使用预先配置的**防火墙流量日志**蓝图。该蓝图包含默认配置，用于检索存储在 Amazon S3 存储桶中的原始安全日志或其他数据、处理这些记录以及将其标准化。随后，该蓝图将数据映射到开放式网络安全模式框架（OCSF），并将转换后的符合 OCSF 标准的数据发送至 Security Lake。 该管道具有以下元数据属性： + `bucket_name`：Security Lake 创建的 Amazon S3 存储桶的名称，用于存储安全数据。 + `path_prefix`：在 Security Lake IAM 角色策略中定义的自定义源名称。 + `region`：Secur AWS 区域 ity Lake S3 存储桶所在的位置。 + `accountID`：启用安全湖的 AWS 账户 ID。 + `sts_role_arn`：用于与 Security Lake 结合使用的 IAM 角色的 ARN。 ## 先决条件 创建管道以将数据发送至 Security Lake 之前，请执行以下步骤： + **启用和配置 Amazon Security Lake**：设置 Amazon Security Lake，以集中来自不同源的安全数据。有关说明，请参阅 [Enabling Security Lake using the console](https://docs.aws.amazon.com/security-lake/latest/userguide/get-started-console.html)。 选择源时，选择**摄取特定 AWS 源**，然后选择一个或多个要摄取的日志和事件来源。 + **设置权限**：配置管道角色所需的权限，以便将数据写入 Security Lake。有关更多信息，请参阅[管道角色](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/pipeline-security-overview.html#pipeline-security-sink)。 ### 创建管道 使用预先配置的 Security Lake 蓝图，以创建管道。有关更多信息，请参阅 [Using blueprints to create a pipeline](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/pipeline-blueprint.html)。 # 将 OpenSearch 采集管道与 Fluent Bit 配合使用 此示例 [Fluent Bit 配置文件](https://docs.fluentbit.io/manual/pipeline/outputs/http)将日志数据从 Fluent Bit 发送到 OpenSearch 摄取管道。有关摄取日志数据的更多信息，请参阅 Data Prepper 文档中的[日志分析](https://github.com/opensearch-project/data-prepper/blob/main/docs/log_analytics.md)。 注意以下几点： + `host` 值必须是管道端点。例如 `pipeline-endpoint.us-east-1osis.amazonaws.com`。 + `aws_service` 值必须为 `osis`。 + 该`aws_role_arn`值是 AWS IAM 角色的 ARN，供客户端代入并用于签名版本 4 身份验证。 ``` [INPUT] name tail refresh_interval 5 path /var/log/test.log read_from_head true [OUTPUT] Name http Match * Host pipeline-endpoint.us-east-1osis.amazonaws.com Port 443 URI /log/ingest Format json aws_auth true aws_region region aws_service osis aws_role_arn arn:aws:iam::account-id:role/ingestion-role Log_Level trace tls On ``` 然后，您可以配置如下所示的 OpenSearch 采集管道，该管道以 HTTP 为源： ``` version: "2" unaggregated-log-pipeline: source: http: path: "/log/ingest" processor: - grok: match: log: - "%{TIMESTAMP_ISO8601:timestamp} %{NOTSPACE:network_node} %{NOTSPACE:network_host} %{IPORHOST:source_ip}:%{NUMBER:source_port:int} -> %{IPORHOST:destination_ip}:%{NUMBER:destination_port:int} %{GREEDYDATA:details}" - grok: match: details: - "'%{NOTSPACE:http_method} %{NOTSPACE:http_uri}' %{NOTSPACE:protocol}" - "TLS%{NOTSPACE:tls_version} %{GREEDYDATA:encryption}" - "%{NUMBER:status_code:int} %{NUMBER:response_size:int}" - delete_entries: with_keys: ["details", "log"] sink: - opensearch: hosts: ["https://search-domain-endpoint.us-east-1es.amazonaws.com"] index: "index_name" index_type: custom bulk_size: 20 aws: region: "region" ``` # 将 OpenSearch 采集管道与 Fluentd 配合使用 Fluentd是一个开源数据收集生态系统，它提供了 SDKs 不同的语言和子项目，例如Fluent Bit。此示例 [Fluentd 配置文件](https://docs.fluentd.org/output/http#example-configuration)将日志数据从 Fluentd 发送到摄取管道。 OpenSearch 有关摄取日志数据的更多信息，请参阅 Data Prepper 文档中的[日志分析](https://github.com/opensearch-project/data-prepper/blob/main/docs/log_analytics.md)。 注意以下几点： + `endpoint` 值必须是管道端点。例如 `pipeline-endpoint.us-east-1osis.amazonaws.com/apache-log-pipeline/logs`。 + `aws_service` 值必须为 `osis`。 + 该`aws_role_arn`值是 AWS IAM 角色的 ARN，供客户端代入并用于签名版本 4 身份验证。 ``` @type tail path logs/sample.log path_key log tag apache @type none @type record_transformer log ${record["message"]} @type record_transformer remove_keys message @type http endpoint pipeline-endpoint.us-east-1osis.amazonaws.com/apache-log-pipeline/logs json_array true method aws_sigv4 aws_service osis aws_region region aws_role_arn arn:aws:iam::account-id:role/ingestion-role @type json flush_interval 1s ``` 然后，您可以配置如下所示的 OpenSearch 采集管道，该管道以 HTTP 为源： ``` version: "2" apache-log-pipeline: source: http: path: "/${pipelineName}/logs" processor: - grok: match: log: - "%{TIMESTAMP_ISO8601:timestamp} %{NOTSPACE:network_node} %{NOTSPACE:network_host} %{IPORHOST:source_ip}:%{NUMBER:source_port:int} -> %{IPORHOST:destination_ip}:%{NUMBER:destination_port:int} %{GREEDYDATA:details}" sink: - opensearch: hosts: ["https://search-domain-endpoint.us-east-1es.amazonaws.com"] index: "index_name" aws_region: "region" aws_sigv4: true ``` # 使用带机器学习离线批量推理的 OpenSearch 摄取管道 Amazon OpenSearch Ingestion (OSI) 管道支持机器学习 (ML) 离线批量推理处理，可以低成本高效地丰富大量数据。如果具有可异步处理的大型数据集，即可使用离线批次推断。离线批量推理适用于 Amazon Bedrock 和模型。 SageMaker 此功能适用于所有支持 S OpenSearch ervice 2. AWS 区域 17\$1 域名的 OpenSearch Ingestion。 **注意** 要进行实时推理处理，请使用 [适用于第三方平台的亚马逊 OpenSearch 服务 ML 连接器](ml-external-connector.md)。 离线批量推理处理利用了 OpenSearch 名为 ML Commons 的功能。*ML Commons* 通过传输和 REST API 调用提供机器学习算法。这些调用为每个机器学习请求选择正确的节点和资源，并监控机器学习任务，以确保正常运行。通过这种方式，ML Commons 使您可以利用现有的开源机器学习算法，减少开发新的机器学习功能所需的工作量。有关 ML Commons 的更多信息，请参阅 OpenSearch .org 文档中的[机器学习](https://docs.opensearch.org/latest/ml-commons-plugin/)。 ## 工作原理 通过向管道[添加机器学习推理处理器，您可以在 In OpenSearch gestion 上创建离线批量推理](https://docs.opensearch.org/latest/ingest-pipelines/processors/ml-inference/)管道。此处理器使您的管道能够连接到 AI 服务 SageMaker ，例如运行批量推理作业。您可以通过在目标域上运行的人工智能连接器（支持 [batch\$1predict](https://docs.opensearch.org/latest/ml-commons-plugin/api/model-apis/batch-predict/)）配置处理器，使其连接到所需的人工智能服务。 OpenSearch Ingestion 使用带有 ML Commons 的`ml_inference`处理器来创建离线批量推理作业。然后，机器学习共享资源使用 [batch\$1predic](https://docs.opensearch.org/latest/ml-commons-plugin/api/model-apis/batch-predict/) t API，该API使用部署在亚马逊 Bedrock、Amazon、 SageMaker Cohere和OpenAI的外部模型服务器上的模型，以离线异步模式对大型数据集进行推理。下图显示了一个 OpenSearch Ingestion 管道，该管道协调多个组件以端到端地执行此过程： ![\[批量 AI 推理处理的三管道架构。\]](http://docs.aws.amazon.com/zh_cn/opensearch-service/latest/developerguide/images/ml_processor.png) 管道组件的工作原理如下： **管道 1（数据准备和转换）\$1：** + 来源：数据是从 OpenSearch Ingestion 支持的外部来源扫描的。 + 数据处理器：原始数据经过处理并转换为正确格式，以便在集成的人工智能服务上进行批次推断。 + S3（接收器）：处理后的数据暂存于 Amazon S3 存储桶中，可随时作为输入数据用于在集成的 AI 服务上运行批次推断任务。 **管道 2（触发 ML batch\$1inference）：** + 来源：通过自动化 S3 事件检测，识别由管道 1 输出创建的新文件。 + Ml\$1inference 处理器：通过异步批处理作业生成 ML 推理的处理器。该处理器通过在目标域上运行的已配置的 AI 连接器连接到 AI 服务。 + 任务 ID：每个批处理作业在 ml-commons 中都关联一个任务 ID，用于跟踪和管理。 + OpenSearch ML Commons：ML Commons 托管实时神经搜索模型，管理与远程 AI 服务器的连接器， APIs 并为批量推理和作业管理提供服务。 + 人工智能服务： OpenSearch 机器学习共享资源与 Amazon Bedrock 和 Amazon 等人工智能服务 SageMaker进行交互，对数据进行批量推断，从而生成预测或见解。结果将异步保存到单独的 S3 文件中。 **管道 3（批量摄取）：** + S3（来源）：批处理作业的结果存储在 S3 中，这是此管道的来源。 + 数据转换处理器：在摄取批次推断输出之前，对其进行进一步的处理和转换。这样可以确保数据在 OpenSearch 索引中正确映射。 + OpenSearch index（Sink）：将处理后的结果编入索引， OpenSearch 以便存储、搜索和进一步分析。 **注意** \$1管道 1 所述流程为可选流程。如果您愿意，可跳过该过程，直接将准备好的数据上传到 S3 接收器，即可创建批处理作业。 ## 关于 ml\$1inference 处理器 OpenSearch Ingestion 使用 S3 扫描源和 ML 推理处理器之间的专门集成进行批处理。S3 Scan 采用纯元数据模式运行，无需读取实际文件内容即可高效收集 S3 文件信息。`ml_inference`处理器使用 S3 文件与 ML 共享资源进行协调 URLs 以进行批处理。此设计通过最大限度地减少扫描阶段不必要的数据传输，优化批次推断工作流程。您可以使用参数定义 `ml_inference` 处理器。示例如下： ``` processor: - ml_inference: # The endpoint URL of your OpenSearch domain host: "https://AWS test-offlinebatch-123456789abcdefg.us-west-2.es.amazonaws.com" # Type of inference operation: # - batch_predict: for batch processing # - predict: for real-time inference action_type: "batch_predict" # Remote ML model service provider (Amazon Bedrock or SageMaker) service_name: "bedrock" # Unique identifier for the ML model model_id: "AWS TestModelID123456789abcde" # S3 path where batch inference results will be stored output_path: "s3://amzn-s3-demo-bucket/" # Supports ISO_8601 notation strings like PT20.345S or PT15M # These settings control how long to keep your inputs in the processor for retry on throttling errors retry_time_window: "PT9M" # AWS configuration settings aws: # AWS 区域 where the Lambda function is deployed region: "us-west-2" # IAM role ARN for Lambda function execution sts_role_arn: "arn:aws::iam::account_id:role/Admin" # Dead-letter queue settings for storing errors dlq: s3: region: us-west-2 bucket: batch-inference-dlq key_path_prefix: bedrock-dlq sts_role_arn: arn:aws:iam::account_id:role/OSI-invoke-ml # Conditional expression that determines when to trigger the processor # In this case, only process when bucket matches "amzn-s3-demo-bucket" ml_when: /bucket == "amzn-s3-demo-bucket" ``` ### 使用 ml\$1inference 处理器提高摄取性能 OpenSearch Ingestion `ml_inference` 处理器显著增强了支持 ML 的搜索的数据摄取性能。对于需要机器学习模型生成数据的使用案例，包括语义搜索、多模态搜索、文档增强和查询理解，该处理器是理想之选。在语义搜索中，处理器能够将大规模、高维向量的创建和摄取速度提升一个数量级。 与实时模型调用相比，处理器的离线批量推断功能具有显著优势。实时处理需要依赖具有容量限制的在线模型服务器，而批次推断则能根据需求动态扩展计算资源，并以并行方式处理数据。例如，当 OpenSearch 摄取管道收到十亿个源数据请求时，它会为机器学习批量推理输入创建 100 个 S3 文件。然后，`ml_inference`处理器使用 100 个 `ml.m4.xlarge` Amazon Elastic Compute Cloud (Amazon EC2) 实例启动 SageMaker 批处理作业，在 14 小时内完成十亿个请求的矢量化——在实时模式下几乎不可能完成这项任务。 ## 配置 ml\$1inference 处理器以接收语义搜索的数据请求 以下过程将引导您完成设置和配置 OpenSearch Ingestion `ml_inference` 处理器的过程，以使用文本嵌入模型提取十亿个用于语义搜索的数据请求。 **Topics** + [步骤 1：创建连接器并在中注册模型 OpenSearch](#configure-clients-ml-commons-configuring-create-connectors) + [步骤 2：为机器学习离线批量推理创建 OpenSearch 摄取管道](#configure-clients-ml-commons-configuring-pipeline) + [步骤 3：准备数据以供摄取](#configure-clients-ml-commons-configuring-data) + [步骤 4：监控批次推断作业](#configure-clients-ml-commons-configuring-monitor) + [步骤 5：运行搜索](#configure-clients-ml-commons-configuring-semantic-search) ### 步骤 1：创建连接器并在中注册模型 OpenSearch 在以下步骤中，使用 ML [Commons batch\$1inference\$1sagemaker\$1connector\$1blueprint](https://github.com/opensearch-project/ml-commons/blob/main/docs/remote_inference_blueprints/batch_inference_sagemaker_connector_blueprint.md) 在亚马逊中创建连接器和模型。 SageMaker如果您更喜欢使用 OpenSearch CloudFormation 集成模板，请参阅本节[（替代程序）步骤 1：使用 CloudFormation 集成模板创建连接器和模型](#configure-clients-ml-commons-configuring-create-connectors-alternative)后面的内容。 **要在中创建连接器并注册模型 OpenSearch** 1. 在中创建深度 Java 库 (DJL) 机器学习模型以 SageMaker 进行批量转换。要查看其他 DJL 模型，请参阅 semantic\$1search\$1with\$1cfn\$1 [tem](https://github.com/opensearch-project/ml-commons/blob/main/docs/tutorials/aws/semantic_search_with_CFN_template_for_Sagemaker.md) plate\$1for\$1sageMaker，网址为： GitHub ``` POST https://api.sagemaker.us-east-1.amazonaws.com/CreateModel { "ExecutionRoleArn": "arn:aws:iam::123456789012:role/aos_ml_invoke_sagemaker", "ModelName": "DJL-Text-Embedding-Model-imageforjsonlines", "PrimaryContainer": { "Environment": { "SERVING_LOAD_MODELS" : "djl://ai.djl.huggingface.pytorch/sentence-transformers/all-MiniLM-L6-v2" }, "Image": "763104351884.dkr.ecr.us-east-1.amazonaws.com/djl-inference:0.29.0-cpu-full" } } ``` 1. 创建连接器，将 `batch_predict` 作为 `actions` 字段中的新 `action` 类型： ``` POST /_plugins/_ml/connectors/_create { "name": "DJL Sagemaker Connector: all-MiniLM-L6-v2", "version": "1", "description": "The connector to sagemaker embedding model all-MiniLM-L6-v2", "protocol": "aws_sigv4", "credential": { "roleArn": "arn:aws:iam::111122223333:role/SageMakerRole" }, "parameters": { "region": "us-east-1", "service_name": "sagemaker", "DataProcessing": { "InputFilter": "$.text", "JoinSource": "Input", "OutputFilter": "$" }, "MaxConcurrentTransforms": 100, "ModelName": "DJL-Text-Embedding-Model-imageforjsonlines", "TransformInput": { "ContentType": "application/json", "DataSource": { "S3DataSource": { "S3DataType": "S3Prefix", "S3Uri": "s3://offlinebatch/msmarcotests/" } }, "SplitType": "Line" }, "TransformJobName": "djl-batch-transform-1-billion", "TransformOutput": { "AssembleWith": "Line", "Accept": "application/json", "S3OutputPath": "s3://offlinebatch/msmarcotestsoutputs/" }, "TransformResources": { "InstanceCount": 100, "InstanceType": "ml.m4.xlarge" }, "BatchStrategy": "SingleRecord" }, "actions": [ { "action_type": "predict", "method": "POST", "headers": { "content-type": "application/json" }, "url": "https://runtime.sagemaker.us-east-1.amazonaws.com/endpoints/OpenSearch-sagemaker-060124023703/invocations", "request_body": "${parameters.input}", "pre_process_function": "connector.pre_process.default.embedding", "post_process_function": "connector.post_process.default.embedding" }, { "action_type": "batch_predict", "method": "POST", "headers": { "content-type": "application/json" }, "url": "https://api.sagemaker.us-east-1.amazonaws.com/CreateTransformJob", "request_body": """{ "BatchStrategy": "${parameters.BatchStrategy}", "ModelName": "${parameters.ModelName}", "DataProcessing" : ${parameters.DataProcessing}, "MaxConcurrentTransforms": ${parameters.MaxConcurrentTransforms}, "TransformInput": ${parameters.TransformInput}, "TransformJobName" : "${parameters.TransformJobName}", "TransformOutput" : ${parameters.TransformOutput}, "TransformResources" : ${parameters.TransformResources}}""" }, { "action_type": "batch_predict_status", "method": "GET", "headers": { "content-type": "application/json" }, "url": "https://api.sagemaker.us-east-1.amazonaws.com/DescribeTransformJob", "request_body": """{ "TransformJobName" : "${parameters.TransformJobName}"}""" }, { "action_type": "cancel_batch_predict", "method": "POST", "headers": { "content-type": "application/json" }, "url": "https://api.sagemaker.us-east-1.amazonaws.com/StopTransformJob", "request_body": """{ "TransformJobName" : "${parameters.TransformJobName}"}""" } ] } ``` 1. 使用返回的连接器 ID 注册 SageMaker 模型： ``` POST /_plugins/_ml/models/_register { "name": "SageMaker model for batch", "function_name": "remote", "description": "test model", "connector_id": "example123456789-abcde" } ``` 1. 使用 `batch_predict` 操作类型调用模型： ``` POST /_plugins/_ml/models/teHr3JABBiEvs-eod7sn/_batch_predict { "parameters": { "TransformJobName": "SM-offline-batch-transform" } } ``` 响应包含批处理作业的任务 ID： ``` { "task_id": "exampleIDabdcefd_1234567", "status": "CREATED" } ``` 1. 使用任务 ID 调用 Get Task API，以检查批处理作业状态： ``` GET /_plugins/_ml/tasks/exampleIDabdcefd_1234567 ``` 响应包含任务状态： ``` { "model_id": "nyWbv5EB_tT1A82ZCu-e", "task_type": "BATCH_PREDICTION", "function_name": "REMOTE", "state": "RUNNING", "input_type": "REMOTE", "worker_node": [ "WDZnIMcbTrGtnR4Lq9jPDw" ], "create_time": 1725496527958, "last_update_time": 1725496527958, "is_async": false, "remote_job": { "TransformResources": { "InstanceCount": 1, "InstanceType": "ml.c5.xlarge" }, "ModelName": "DJL-Text-Embedding-Model-imageforjsonlines", "TransformOutput": { "Accept": "application/json", "AssembleWith": "Line", "KmsKeyId": "", "S3OutputPath": "s3://offlinebatch/output" }, "CreationTime": 1725496531.935, "TransformInput": { "CompressionType": "None", "ContentType": "application/json", "DataSource": { "S3DataSource": { "S3DataType": "S3Prefix", "S3Uri": "s3://offlinebatch/sagemaker_djl_batch_input.json" } }, "SplitType": "Line" }, "TransformJobArn": "arn:aws:sagemaker:us-east-1:111122223333:transform-job/SM-offline-batch-transform15", "TransformJobStatus": "InProgress", "BatchStrategy": "SingleRecord", "TransformJobName": "SM-offline-batch-transform15", "DataProcessing": { "InputFilter": "$.content", "JoinSource": "Input", "OutputFilter": "$" } } } ``` #### （替代程序）步骤 1：使用 CloudFormation 集成模板创建连接器和模型 如果您愿意，可以使用 AWS CloudFormation 自动创建所有必需的 Amazon SageMaker 连接器和模型以进行机器学习推理。此方法通过使用 Amazon S OpenSearch ervice 控制台中提供的预配置模板来简化设置。有关更多信息，请参阅 [用于 CloudFormation 为语义搜索设置远程推理](cfn-template.md)。 **部署可创建所有必需 SageMaker 连接器和模型的 CloudFormation 堆栈** 1. 打开亚马逊 OpenSearch 服务控制台。 1. 在导航窗格中，选择**集成**。 1. 在 “搜索” 字段中输入**SageMaker**，然后选择 “**通过 Amazon 与文本嵌入模型集成**” SageMaker。 1. 选择**配置域**，然后选择**配置 VPC 域**或**配置公共域**。 1. 在模板字段中输入信息。对于**启用离线批次推断**，选择 **true**，以配置用于离线批量处理的资源。 1. 选择 “**创建**” 以创建 CloudFormation 堆栈。 1. 创建堆栈后，在 CloudFormation 控制台中打开**输出**选项卡，查找 **connector\$1id** 和 **model\$1id**。在后续配置管道时，您将需要这些值。 ### 步骤 2：为机器学习离线批量推理创建 OpenSearch 摄取管道 使用以下示例为机器学习离线批量推 OpenSearch 断创建采集管道。有关为 OpenSearch Ingestion 创建管道的更多信息，请参阅。[创建 Amazon OpenSearch Ingestion 管道](creating-pipeline.md) **开始之前** 在以下示例中，您可以为 `sts_role_arn` 参数指定 IAM 角色 ARN。使用以下步骤验证此角色是否已映射到有权访问 ml-commons 的后端角色。 OpenSearch 1. 导航到您的 OpenSearch 服务域的 OpenSearch 仪表板插件。您可以在 OpenSearch 服务控制台的域控制面板上找到仪表板终端节点。 1. 从主菜单中选择**安全**、**角色**，然后选择 **ml\$1full\$1access** 角色。 1. 选择**映射的用户**、**管理映射**。 1. 在**后端角色**下，输入需要权限才能调用域的 Lambda 角色的 ARN。这是一个例子：arn: aws: iam::: role/ *111122223333* *lambda-role* 1. 选择**映射**并确认在**映射的用户**下显示的用户或角色。 **为机器学习离线批量推理创建 OpenSearch 摄取管道的示例** ``` version: '2' extension: osis_configuration_metadata: builder_type: visual sagemaker-batch-job-pipeline: source: s3: acknowledgments: true delete_s3_objects_on_read: false scan: buckets: - bucket: name: name data_selection: metadata_only filter: include_prefix: - sagemaker/sagemaker_djl_batch_input exclude_suffix: - .manifest - bucket: name: name data_selection: data_only filter: include_prefix: - sagemaker/output/ scheduling: interval: PT6M aws: region: name default_bucket_owner: account_ID codec: ndjson: include_empty_objects: false compression: none workers: '1' processor: - ml_inference: host: "https://search-AWStest-offlinebatch-123456789abcdef.us-west-2.es.amazonaws.com" aws_sigv4: true action_type: "batch_predict" service_name: "sagemaker" model_id: "model_ID" output_path: "s3://AWStest-offlinebatch/sagemaker/output" aws: region: "us-west-2" sts_role_arn: "arn:aws:iam::account_ID:role/Admin" ml_when: /bucket == "AWStest-offlinebatch" dlq: s3: region: us-west-2 bucket: batch-inference-dlq key_path_prefix: bedrock-dlq sts_role_arn: arn:aws:iam::account_ID:role/OSI-invoke-ml - copy_values: entries: - from_key: /text to_key: chapter - from_key: /SageMakerOutput to_key: chapter_embedding - delete_entries: with_keys: - text - SageMakerOutput sink: - opensearch: hosts: ["https://search-AWStest-offlinebatch-123456789abcdef.us-west-2.es.amazonaws.com"] aws: serverless: false region: us-west-2 routes: - ml-ingest-route index_type: custom index: test-nlp-index routes: - ml-ingest-route: /chapter != null and /title != null ``` ### 步骤 3：准备数据以供摄取 要为机器学习离线批量推理处理准备数据，请使用自己的工具或流程自行准备数据，也可以使用 [OpenSearch Data Prepper](https://docs.opensearch.org/latest/data-prepper/getting-started/)。确认数据已按正确格式组织，具体可通过两种方式实现：使用管道从数据来源获取数据，或创建机器学习数据集。 以下示例使用 [MS MARCO](https://microsoft.github.io/msmarco/Datasets.html) 数据集，该数据集包含一组用于自然语言处理任务的真实用户查询。该数据集采用 JSONL 格式，其中每行代表向机器学习嵌入模型发送的一个请求： ``` {"_id": "1185869", "text": ")what was the immediate impact of the Paris Peace Treaties of 1947?", "metadata": {"world war 2"}} {"_id": "1185868", "text": "_________ justice is designed to repair the harm to victim, the community and the offender caused by the offender criminal act. question 19 options:", "metadata": {"law"}} {"_id": "597651", "text": "what is amber", "metadata": {"nothing"}} {"_id": "403613", "text": "is autoimmune hepatitis a bile acid synthesis disorder", "metadata": {"self immune"}} ... ``` 要使用 MS MARCO 数据集进行测试，请设想这样的场景：构建 10 亿个输入请求，这些请求分布在 100 个文件中，每个文件包含 1000 万个请求。这些文件将存储在亚马逊 S3 中，前缀为 s3://offlinebatch/sagemaker/sagemaker\$1djl\$1batch\$1input/。 OpenSearch Ingestion 管道将同时扫描这 100 个文件，并启动一个包含 100 个工作人员的 SageMaker 批处理作业进行并行处理，从而实现高效的矢量化处理，并将这十亿个文档摄入其中。 OpenSearch 在生产环境中，您可以使用 OpenSearch 摄取管道为批量推理输入生成 S3 文件。该管道支持多种[数据来源](https://docs.opensearch.org/latest/data-prepper/pipelines/configuration/sources/sources/)，并按计划运行，持续将源数据转换为 S3 文件。随后，人工智能服务器会通过计划的离线批处理作业自动处理这些文件，从而确保持续的数据处理和摄取。 ### 步骤 4：监控批次推断作业 您可以使用 SageMaker 控制台或监控批量推理作业。 AWS CLI您还可以使用 Get Task API 监控批量作业： ``` GET /_plugins/_ml/tasks/_search { "query": { "bool": { "filter": [ { "term": { "state": "RUNNING" } } ] } }, "_source": ["model_id", "state", "task_type", "create_time", "last_update_time"] } ``` 该 API 会返回活跃批处理作业任务的列表： ``` { "took": 2, "timed_out": false, "_shards": { "total": 5, "successful": 5, "skipped": 0, "failed": 0 }, "hits": { "total": { "value": 3, "relation": "eq" }, "max_score": 0.0, "hits": [ { "_index": ".plugins-ml-task", "_id": "nyWbv5EB_tT1A82ZCu-e", "_score": 0.0, "_source": { "model_id": "nyWbv5EB_tT1A82ZCu-e", "state": "RUNNING", "task_type": "BATCH_PREDICTION", "create_time": 1725496527958, "last_update_time": 1725496527958 } }, { "_index": ".plugins-ml-task", "_id": "miKbv5EB_tT1A82ZCu-f", "_score": 0.0, "_source": { "model_id": "miKbv5EB_tT1A82ZCu-f", "state": "RUNNING", "task_type": "BATCH_PREDICTION", "create_time": 1725496528123, "last_update_time": 1725496528123 } }, { "_index": ".plugins-ml-task", "_id": "kiLbv5EB_tT1A82ZCu-g", "_score": 0.0, "_source": { "model_id": "kiLbv5EB_tT1A82ZCu-g", "state": "RUNNING", "task_type": "BATCH_PREDICTION", "create_time": 1725496529456, "last_update_time": 1725496529456 } } ] } } ``` ### 步骤 5：运行搜索 监控批次推断作业并确认其完成后，您可以运行多种类型的 AI 搜索，包括语义搜索、混合搜索、对话式搜索（使用 RAG）、神经稀疏搜索和多模态搜索。有关 OpenSearch 服务支持的 AI 搜索的更多信息，请参阅 [AI 搜索](https://docs.opensearch.org/latest/vector-search/ai-search/index/)。 要搜索原始向量，请使用 `knn` 查询类型，提供 `vector` 数组作为输入，并指定返回的结果数量 `k`： ``` GET /my-raw-vector-index/_search { "query": { "knn": { "my_vector": { "vector": [0.1, 0.2, 0.3], "k": 2 } } } } ``` 要运行 AI 驱动的搜索，请使用 `neural` 查询类型。指定`query_text`输入、您在 `model_id` OpenSearch Ingestion 管道中配置的嵌入模型以及返回的结果`k`数。要将嵌入字段排除在搜索结果之外，请在 `_source.excludes` 参数中指定嵌入字段的名称： ``` GET /my-ai-search-index/_search { "_source": { "excludes": [ "output_embedding" ] }, "query": { "neural": { "output_embedding": { "query_text": "What is AI search?", "model_id": "mBGzipQB2gmRjlv_dOoB", "k": 2 } } } } ``` # 将采集管道与 C OpenSearch ollector 配合使用 OpenTelemetry 您可以使用[OpenTelemetry 收集器](https://opentelemetry.io/docs/collector/)将日志、跟踪和指标提取到 OpenSearch 摄取管道中。单个管道可用于将所有日志、跟踪和指标提取到域或集合上的不同索引。您还可以使用管道仅单独采集日志、跟踪或指标。 **Topics** + [先决条件](#otel-prereqs) + [步骤 1：配置管道角色](#otel-pipeline-role) + [步骤 2：创建管道](#create-otel-pipeline) + [跨账户连接](#x-account-connectivity) + [限制](#otel-limitations) + [为 OpenTelemetry 信号源推荐的 CloudWatch 警报](#otel-pipeline-metrics) ## 先决条件 在设置[OpenTelemetry 配置文件](https://opentelemetry.io/docs/collector/configuration/)时，必须配置以下内容才能进行摄取： + 摄取角色需要`osis:Ingest`权限才能与管道交互。有关更多信息，请参阅[摄取角色。](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/pipeline-security-overview.html#pipeline-security-same-account) + 端点值必须包含您的管道端点。例如，`https://pipeline-endpoint.us-east-1.osis.amazonaws.com.` + 服务值必须为`osis`。 + OTLP/HTTP 导出器的压缩选项必须与管道所选源上的压缩选项相匹配。 ``` extensions: sigv4auth: region: "region" service: "osis" exporters: otlphttp: logs_endpoint: "https://pipeline-endpoint.us-east-1.osis.amazonaws.com/v1/logs" metrics_endpoint: "https://pipeline-endpoint.us-east-1.osis.amazonaws.com/v1/metrics" traces_endpoint: "https://pipeline-endpoint.us-east-1.osis.amazonaws.com/v1/traces" auth: authenticator: sigv4auth compression: none service: extensions: [sigv4auth] pipelines: traces: receivers: [jaeger] exporters: [otlphttp] ``` ## 步骤 1：配置管道角色 设置 OpenTelemetry 收集器配置后，[设置要在工作流配置中使用的管道角色](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/pipeline-security-overview.html#pipeline-security-sink)。管道角色不需要 OTLP 源所需的特定权限，只有授予管道访问 OpenSearch 域或集合的权限。 ## 步骤 2：创建管道 然后，您可以配置如下所示的 OpenSearch Ingestion 管道，将 OTLP 指定为来源。您还可以将 OpenTelemetry 日志、指标和跟踪配置为单独的来源。 OTLP 源管道配置： ``` version: 2 otlp-pipeline: source: otlp: logs_path: /otlp-pipeline/v1/logs traces_path: /otlp-pipeline/v1/traces metrics_path: /otlp-pipeline/v1/metrics sink: - opensearch: hosts: ["https://search-mydomain.region.es.amazonaws.com"] index: "ss4o_metrics-otel-%{yyyy.MM.dd}" index_type: custom aws: region: "region" ``` OpenTelemetry 日志管道配置： ``` version: 2 otel-logs-pipeline: source: otel_logs_source: path: /otel-logs-pipeline/v1/logs sink: - opensearch: hosts: ["https://search-mydomain.region.es.amazonaws.com"] index: "ss4o_metrics-otel-%{yyyy.MM.dd}" index_type: custom aws: region: "region" ``` OpenTelemetry 指标管道配置： ``` version: 2 otel-metrics-pipeline: source: otel_metrics_source: path: /otel-metrics-pipeline/v1/metrics sink: - opensearch: hosts: ["https://search-mydomain.region.es.amazonaws.com"] index: "ss4o_metrics-otel-%{yyyy.MM.dd}" index_type: custom aws: region: "region" ``` OpenTelemetry 跟踪管道配置： ``` version: 2 otel-trace-pipeline: source: otel_trace_source: path: /otel-traces-pipeline/v1/traces sink: - opensearch: hosts: ["https://search-mydomain.region.es.amazonaws.com"] index: "ss4o_metrics-otel-%{yyyy.MM.dd}" index_type: custom aws: region: "region" ``` 您可以使用预先配置的蓝图，以创建此管道。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 ## 跨账户连接 OpenSearch 具有 OpenTelemetry 来源的摄取管道具有跨账户摄取功能。Amazon OpenSearch Ingestion 允许您共享 AWS 账户 从虚拟私有云 (VPC) 到独立 VPC 中的管道终端节点的管道。有关更多信息，请参阅 [为跨 OpenSearch 账户摄取配置摄取管道](cross-account-pipelines.md)。 ## 限制 OpenSearch 摄取管道无法接收任何大于 20mb 的请求。此值由用户在`max_request_length`选项中配置。此选项默认为 10mb。 ## 为 OpenTelemetry 信号源推荐的 CloudWatch 警报 建议使用以下 CloudWatch 指标来监控您的摄取管道的性能。这些指标可能有助您确定处理的导出数据量、处理的流事件量、处理导出和流事件时的错误数以及写入目标的文档数量。您可以设置 CloudWatch 警报，以便在其中一个指标在指定时间内超过指定值时执行操作。 OTLP 来源的 CloudWatch 指标格式为。`{pipeline-name}.otlp.{logs | traces | metrics}.{metric-name}`例如 `otel-pipeline.otlp.metrics.requestTimeouts.count`。 如果使用单个 OpenTelemetry 来源，则指标的格式将为`{pipeline-name}.{source-name}.{metric-name}`。例如 `trace-pipeline.otel_trace_source.requestTimeouts.count`。 所有三种 OpenTelemetry 数据类型都将具有相同的指标，但为简洁起见，下表中仅列出了 OTLP 源日志类型数据的指标。 | 指标 | 说明 | | --- |--- | | otel-pipeline.BlockingBuffer.bufferUsage.value | 指示正在使用的缓冲区数量。 | | otel-pipeline.otlp.logs.requestTimeouts.count | 已超时的请求数。 | | otel-pipeline.otlp.logs.requestsReceived.count | OpenTelemetry 收集器收到的请求数。 | | otel-pipeline.otlp.logs.badRequests.count | OpenTelemetry 收集器收到的格式错误的请求数。 | | otel-pipeline.otlp.logs.requestsTooLarge.count | OpenTelemetry 收集器收到的大于最大值 20mb 的请求数。 | | otel-pipeline.otlp.logs.internalServerError.count | 从 OpenTelemetry 收集器收到的 HTTP 500 错误数。 | | otel-pipeline.opensearch.bulkBadRequestErrors.count | 批量请求期间由于请求格式错误而导致的错误计数。 | | otel-pipeline.opensearch.bulkRequestLatency.avg | 向发出的批量写入请求的平均延迟 OpenSearch。 | | otel-pipeline.opensearch.bulkRequestNotFoundErrors.count | 由于找不到目标数据而失败的批量请求数。 | | otel-pipeline.opensearch.bulkRequestNumberOfRetries.count | 采集管道重试写 OpenSearch 入集群的次数。 OpenSearch | | otel-pipeline.opensearch.bulkRequestSizeBytes.sum | 向发出的所有批量请求的总大小（以字节为单位） OpenSearch。 | | otel-pipeline.opensearch.documentErrors.count | 向发送文档时出现的错误数 OpenSearch。导致错误的文件将发送给 DLQ。 | | otel-pipeline.opensearch.documentsSuccess.count | 成功写入 OpenSearch 集群或集合的文档数。 | | otel-pipeline.opensearch.documentsSuccessFirstAttempt.count | 第一次尝试成功编入索引 OpenSearch 的文档数。 | | `otel-pipeline.opensearch.documentsVersionConflictErrors.count` | 处理过程中由于文档版本冲突而导致的错误计数。 | | `otel-pipeline.opensearch.PipelineLatency.avg` | OpenSearch Ingestion 管道通过从源读取数据到写入目标来处理数据的平均延迟。 | | otel-pipeline.opensearch.PipelineLatency.max | 通过从源读取数据到写入目标来处理数据的 OpenSearch Ingestion 管道的最大延迟。 | | otel-pipeline.opensearch.recordsIn.count | 成功摄入 OpenSearch的记录数。该指标对于跟踪正在处理和存储的数据量至关重要。 | | otel-pipeline.opensearch.s3.dlqS3RecordsFailed.count | 未能写入 DLQ 的记录数。 | | otel-pipeline.opensearch.s3.dlqS3RecordsSuccess.count | 写入 DLQ 的记录数。 | | otel-pipeline.opensearch.s3.dlqS3RequestLatency.count | Amazon S3 死信队列请求的延迟测量次数。 | | otel-pipeline.opensearch.s3.dlqS3RequestLatency.sum | Amazon S3 死信队列的所有请求的总延迟 | | otel-pipeline.opensearch.s3.dlqS3RequestSizeBytes.sum | 向 Amazon S3 死信队列发出的所有请求的总大小（以字节为单位）。 | | otel-pipeline.recordsProcessed.count | 管道中处理的记录总数，这是衡量整体吞吐量的关键指标。 | | `otel-pipeline.opensearch.bulkRequestInvalidInputErrors.count` | OpenSearch 由于输入无效而导致的批量请求错误计数，这对于监控数据质量和操作问题至关重要。 | # 在适用于 Pro OpenSearch metheus 的亚马逊托管服务中使用采集管道 您可以使用适用于 Prometheus 的亚马逊托管服务作为采集渠道的目的地，以 OpenSearch 时间序列格式存储指标。Prometheus 接收器允许您将管道中的指标或其他时间序列数据 OpenTelemetry 发送到适用于 Prometheus 的亚马逊托管服务工作区，以进行监控、提醒和分析。 通过`prometheus`接收器插件， OpenSearch Ingestion 管道能够使用 Prometheus 远程写入协议将指标数据写入适用于 Prometheus 工作空间的亚马逊托管服务。此集成使您能够： + 将时间序列指标数据存储在适用于 Prometheus 的亚马逊托管服务中 + 使用适用于 Prometheus 的亚马逊托管服务和亚马逊托管 Grafana 监控指标并发出警报 + 同时将指标路由到多个目的地（例如， OpenSearch 以及适用于 Prometheus 的亚马逊托管服务） + 处理来自外部代理的 OpenTelemetry 指标或在管道内生成指标 **Topics** + [先决条件](#prometheus-prereqs) + [步骤 1：配置管道角色](#prometheus-pipeline-role) + [步骤 2：创建管道](#prometheus-pipeline) + [监控和排查](#prometheus-monitoring) + [限制](#prometheus-limitations) + [最佳实践](#prometheus-best-practices) ## 先决条件 在配置 Prometheus 接收器之前，请确保具备以下条件： + **适用于 Prometheus 的亚马逊托管服务**工作空间：在与您的摄取管道 AWS 账户 相同 AWS 区域 的工作空间中创建一个工作空间。 OpenSearch 有关说明，请参阅《适用于 *Prometheus 的亚马逊托管服务用户*指南》中的[创建工作空间](https://docs.aws.amazon.com/prometheus/latest/userguide/AMP-onboard-create-workspace.html)。 + **IAM 权限**：将 IAM 角色配置为具有写入适用于 Prometheus 的亚马逊托管服务的权限。有关更多信息，请参阅 [步骤 1：配置管道角色](#prometheus-pipeline-role)。 **注意** 适用于 Prometheus 工作空间的亚马逊托管服务必须使用服务托管密钥。 AWS AWS KMS Amazon Prometheus 托管服务目前不支持客户管理的 AWS KMS 密钥 Prometheus 接收管道中的接收渠道。 OpenSearch ## 步骤 1：配置管道角色 Prometheus 接收器会自动继承[管道角色](pipeline-security-overview.md#pipeline-security-sink)的 IAM 身份验证权限，因此无需在接收器设置中进行其他角色配置（`sts_role_arn`例如）。 以下示例策略显示了使用适用于 Prometheus 的亚马逊托管服务作为接收器所需的权限： ``` { "Version": "2012-10-17", "Statement": [ { "Sid": "AMPRemoteWrite", "Effect": "Allow", "Action": [ "aps:RemoteWrite" ], "Resource": "arn:aws:aps:region:account-id:workspace/workspace-id" } ] } ``` 替换以下占位符： + `region`: 你的 AWS 区域 （例如，`us-east-1`） + `account-id`: 你的 AWS 账户 身份证 + `workspace-id`: 您的亚马逊托管服务 Prometheus 工作空间 ID 您必须将这些权限附加到您的管道角色。 确保您的管道角色具有允许 OpenSearch Ingestion担任该角色的信任关系： ``` { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "osis-pipelines.amazonaws.com" }, "Action": "sts:AssumeRole" } ] } ``` ## 步骤 2：创建管道 设置权限后，您可以将 OpenSearch 采集管道配置为使用适用于 Prometheus 的亚马逊托管服务作为接收器。 ### 基本配置 以下示例显示了 Prometheus 接收器的最小配置： ``` version: "2" sink: - prometheus: url: "https://aps-workspaces.region.amazonaws.com/workspaces/workspace-id/api/v1/remote_write" aws: region: "region" ``` 您必须在`prometheus`接收器配置中指定`url`选项，即适用于 Prometheus 的亚马逊托管服务远程写入端点。要格式化 URL，请在适用于 Prometheus 的亚马逊托管服务控制台中找到您的工作空间 ID，然后按如下方式构建 URL：。`https://aps-workspaces.region.amazonaws.com/workspaces/workspace-id/api/v1/remote_write` ### 配置选项 使用以下选项为 Prometheus 接收器配置批处理和刷新行为： **Prometheus 水槽配置选项** | Option | 必需 | Type | 说明 | | --- | --- | --- | --- | | max\$1events | 否 | 整数 | 在刷新到 Prometheus 之前要累积的最大事件数。默认值为 1000。 | | max\$1request\$1size | 否 | 字节数 | 刷新前请求负载的最大大小。默认值为 1mb。 | | flush\$1interval | 否 | Duration | 在刷新事件之前等待的最长时间。默认值为 10s。允许的最大值为60s。 | ### 管道示例 **示例 1：适用于 Prometheus 的亚马逊托管服务的 OpenTelemetry 指标** 此管道接收来自外部代理的 OpenTelemetry 指标，并将其写入适用于 Prometheus 的亚马逊托管服务： ``` version: "2" source: otel_metrics_source: path: "/v1/metrics" output_format: otel sink: - prometheus: url: "https://aps-workspaces.us-east-1.amazonaws.com/workspaces/ws-a1b2c3d4-5678-90ab-cdef-EXAMPLE11111/api/v1/remote_write" aws: region: "us-east-1" ``` **示例 2：双水槽—— OpenSearch 以及适用于 Prometheus 的亚马逊托管服务** 此渠道将指标传送到两者 OpenSearch 以及适用于 Prometheus 的亚马逊托管服务： ``` version: "2" source: otel_metrics_source: path: "/v1/metrics" output_format: otel sink: - opensearch: hosts: - "https://search-domain-endpoint.us-east-1.es.amazonaws.com" index: "metrics-%{yyyy.MM.dd}" aws: region: "us-east-1" sts_role_arn: "arn:aws:iam::123456789012:role/OSI-Pipeline-Role" - prometheus: url: "https://aps-workspaces.us-east-1.amazonaws.com/workspaces/ws-a1b2c3d4-5678-90ab-cdef-EXAMPLE11111/api/v1/remote_write" aws: region: "us-east-1" ``` **示例 3：带筛选功能的指标** 在将指标发送到亚马逊 Prometheus 托管服务之前，此渠道会筛选指标： ``` version: "2" source: otel_metrics_source: path: "/v1/metrics" output_format: otel processor: - drop_events: drop_when: '/name != "http.server.duration" and /name != "http.client.duration"' sink: - prometheus: url: "https://aps-workspaces.us-east-1.amazonaws.com/workspaces/ws-a1b2c3d4-5678-90ab-cdef-EXAMPLE11111/api/v1/remote_write" aws: region: "us-east-1" ``` 您可以使用预先配置的 Amazon Prometheus 托管服务蓝图来创建这些管道。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 ### 使用亚马逊托管服务为 Prometheus sink 创建管道 #### 使用控制 AWS 台 1. 导航到 OpenSearch 服务控制台。 1. 在 “接入” 下选择 “****管**道**”。 1. 选择 **Create pipeline（创建管道）**。 1. 选择 “**使用蓝图构建**”，然后选择 **Amazon Prometheus 蓝图的OpenTelemetry 指标**。 1. 配置管道： + 输入你的 Prometheus 亚马逊托管服务工作空间 ID + 指定管道角色 ARN + 根据需要配置源和处理器设置 1. 查看并创建管道。 #### 使用 AWS CLI 使用所需的配置创建工作流配置文件（例如`amp-pipeline.yaml`），然后运行： ``` aws osis create-pipeline \ --pipeline-name my-amp-pipeline \ --min-units 2 \ --max-units 4 \ --pipeline-configuration-body file://amp-pipeline.yaml ``` #### 使用 AWS CloudFormation ``` Resources: MyAMPPipeline: Type: AWS::OSIS::Pipeline Properties: PipelineName: my-amp-pipeline MinUnits: 2 MaxUnits: 4 PipelineConfigurationBody: | version: "2" source: otel_metrics_source: path: "/v1/metrics" output_format: otel sink: - prometheus: url: "https://aps-workspaces.us-east-1.amazonaws.com/workspaces/ws-a1b2c3d4-5678-90ab-cdef-EXAMPLE11111/api/v1/remote_write" aws: region: "us-east-1" ``` ## 监控和排查 ### CloudWatch 指标 使用 CloudWatch 指标监控管道的绩效： + `DocumentsWritten`: 成功写入亚马逊 Prometheus 托管服务的指标数量 + `DocumentsWriteFailed`: 写入失败的指标数量 + `RequestLatency`: 远程写入请求的延迟 ### 常见问题 **问题**：Pipeline 无法写信给 Prometheus 的亚马逊托管服务 **解决方案** + 验证 URL 中的工作空间 ID 和区域是否正确 + 确保管道角色拥有`aps:RemoteWrite`权限 + 检查工作空间是否使用服务管理 AWS KMS 的密钥 + 验证管道和工作空间是否相同 AWS 账户 **问题**：身份验证错误 **解决方案** + 验证信任关系是否`osis-pipelines.amazonaws.com`允许担任管道角色 + 确保管道角色具有所需的`aps:RemoteWrite`权限 **问题**：高延迟或限流 **解决方案** + 增加管道容量单位 + 在处理器中实现批处理 + 查看亚马逊托管服务 Prometheus 服务配额 ## 限制 为适用于 Prometheus 的亚马逊托管服务设置 OpenSearch 摄取管道时，请考虑以下限制： + 适用于 Prometheus 工作空间的亚马逊托管服务必须使用服务托管密钥。 AWS AWS KMS 目前不支持客户管理的 AWS KMS 密钥。 + Pipeline 和 Amazon Prometheus 托管服务工作空间必须相同。 AWS 账户 ## 最佳实践 + **使用相同的 IAM 角色**：Prometheus 接收器会自动使用管道角色。如果使用其他水槽`sts_role_arn`，请确保与管道角色相同 + **监控指标**：为写入失败和高延迟设置 CloudWatch 警报 + **实施筛选**：在发送到亚马逊 Prometheus 托管服务之前，使用处理器筛选不必要的指标 + **适当调整容量**：从最小容量开始，然后根据指标容量进行扩展 + **使用蓝图**：将预先配置的蓝图用于常见用例 # 在 Kafka 中使用 OpenSearch 采集管道 您可以使用 [Kafka](https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sources/kafka/) 插件将数据从自行管理的 Kafka 集群流式传输到亚马逊 OpenSearch 服务域和 OpenSearch 无服务器集合。 OpenSearch Ingestion 支持来自配置有公共或私有 (VPC) 网络的 Kafka 集群的连接。本主题概述设置摄取管道的先决条件和步骤，包括配置网络设置和身份验证方法，例如双向TLS（mTLS）、SASL/SCRAM 或 IAM。 ## 从公有 Kafka 集群迁移数据 您可以使用 OpenSearch Ingestion 管道从公共自管 Kafka 集群迁移数据，这意味着可以公开解析域 DNS 名称。为此，请设置一个以自管理的 Kafka 为源，将 OpenSearch 服务或 OpenSearch 无服务器作为目标的 OpenSearch 摄取管道。这将处理您从自行管理的源集群到托 AWS管目标域或集合的流式数据。 ### 先决条件 在创建 OpenSearch 摄取管道之前，请执行以下步骤： 1. 使用公有网络配置创建自主管理型 Kafka 集群。集群应包含您要采集到 S OpenSearch ervice 中的数据。 1. 创建要将数据迁移到的 OpenSearch 服务域或 OpenSearch 无服务器集合。有关更多信息，请参阅[创建 OpenSearch 服务域](createupdatedomains.md#createdomains)和[创建集合](serverless-create.md)。 1. 使用在您的自管理集群上设置身份验证。 AWS Secrets Manager按照 [Rotate AWS Secrets Manager secrets](https://docs.aws.amazon.com/secretsmanager/latest/userguide/rotating-secrets.html) 中的步骤启用密钥轮换。 1. 将[基于资源的策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/ac.html#ac-types-resource)附加到域，或将[数据访问策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/serverless-data-access.html)附加到集合。这些访问策略允许 OpenSearch Ingestion 将数据从您的自管理集群写入您的域或集合。 以下示例域访问策略允许您在下一步中创建的管道角色将数据写入域。确保使用自身 ARN 更新 `resource`。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "AWS": "arn:aws:iam::444455556666:role/pipeline-role" }, "Action": [ "es:DescribeDomain", "es:ESHttp*" ], "Resource": [ "arn:aws:es:us-east-1:111122223333:domain/domain-name" ] } ] } ``` ------ 要创建具有向该集合或域写入数据的正确权限的 IAM 角色，请参阅 [在 Amazon OpenSearch Ingestion 中设置角色和用户](pipeline-security-overview.md)。 ### 步骤 1：配置管道角色 设置 Kafka 管道先决条件后，[配置要在工作流配置中使用的管道角色](pipeline-security-overview.md#pipeline-security-sink)，添加写入 OpenSearch 服务域或 OpenSearch 无服务器集合的权限，以及从 Secrets Manager 读取密钥的权限。 ### 步骤 2：创建管道 然后，您可以配置如下所示的 OpenSearch Ingestion 管道，将 Kafka 指定为来源。 您可以将多个 OpenSearch 服务域指定为数据的目的地。此功能允许有条件地路由或将传入数据复制到多个 OpenSearch 服务域。 您还可以将数据从源 Confluent Kafka 集群迁移到无服务器 V OpenSearch PC 集合。务必要在管道配置中提供网络访问策略。您可以使用 Confluent 架构注册表来定义 Confluent 架构。 ``` version: "2" kafka-pipeline: source: kafka: encryption: type: "ssl" topics: - name: "topic-name" group_id: "group-id" bootstrap_servers: - "bootstrap-server.us-east-1.aws.private.confluent.cloud:9092" authentication: sasl: plain: username: ${aws_secrets:confluent-kafka-secret:username} password: ${aws_secrets:confluent-kafka-secret:password} schema: type: confluent registry_url: https://my-registry.us-east-1.aws.confluent.cloud api_key: "${{aws_secrets:schema-secret:schema_registry_api_key}}" api_secret: "${{aws_secrets:schema-secret:schema_registry_api_secret}}" basic_auth_credentials_source: "USER_INFO" sink: - opensearch: hosts: ["https://search-mydomain.us-east-1.es.amazonaws.com"] aws: region: "us-east-1" index: "confluent-index" extension: aws: secrets: confluent-kafka-secret: secret_id: "my-kafka-secret" region: "us-east-1" schema-secret: secret_id: "my-self-managed-kafka-schema" region: "us-east-1" ``` 您可以使用预先配置的蓝图，以创建此管道。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 ### 从 VPC 中的 Kafka 集群迁移数据 您还可以使用 OpenSearch 摄取管道从 VPC 中运行的自管 Kafka 集群迁移数据。为此，请设置一个以自管理的 Kafka 为源，将 OpenSearch 服务或 OpenSearch 无服务器作为目标的 OpenSearch 摄取管道。这将处理您从自行管理的源集群到托 AWS管目标域或集合的流式数据。 #### 先决条件 在创建 OpenSearch 摄取管道之前，请执行以下步骤： 1. 使用 VPC 网络配置创建自管理 Kafka 集群，其中包含您要提取到服务中的数据。 OpenSearch 1. 创建要将数据迁移到的 OpenSearch 服务域或 OpenSearch 无服务器集合。有关更多信息，请参阅[创建 OpenSearch 服务域](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/createupdatedomains.html#createdomains)和[创建集合](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/serverless-manage.html#serverless-create)。 1. 使用在您的自管理集群上设置身份验证。 AWS Secrets Manager按照 [Rotate AWS Secrets Manager secrets](https://docs.aws.amazon.com/secretsmanager/latest/userguide/rotating-secrets.html) 中的步骤启用密钥轮换。 1. 获取有权访问自主管理型 Kafka 的 VPC 的 ID。选择 OpenSearch Ingestion 要使用的 VPC CIDR。 **注意** 如果您使用创建管道，则还必须将您的 OpenSearch 摄取管道连接到您的 VPC 才能使用自我管理的 Kafka。 AWS 管理控制台 要执行此操作，请找到**网络配置**部分，选择**连接到 VPC** 复选框，然后从提供的任意一个默认选项中选择 CIDR，或者选择自己的 CIDR。您可以按照 [RFC 1918 当前最佳实践](https://datatracker.ietf.org/doc/html/rfc1918)中的定义，使用私有地址空间中的任何 CIDR。 要提供自定义 CIDR，请从下拉菜单中选择**其他**。为避免接入和自行管理之间的 IP 地址冲突 OpenSearch，请确保自管理 VP OpenSearch C CIDR 与用于 OpenSearch 摄取的 CIDR 不同。 OpenSearch 1. 将[基于资源的策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/ac.html#ac-types-resource)附加到域，或将[数据访问策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/serverless-data-access.html)附加到集合。这些访问策略允许 OpenSearch Ingestion 将数据从您的自管理集群写入您的域或集合。 以下示例域访问策略允许您在下一步中创建的管道角色将数据写入域。确保使用自身 ARN 更新 `resource`。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "AWS": "arn:aws:iam::444455556666:role/pipeline-role" }, "Action": [ "es:DescribeDomain", "es:ESHttp*" ], "Resource": [ "arn:aws:es:us-east-1:111122223333:domain/domain-name" ] } ] } ``` ------ 要创建具有向该集合或域写入数据的正确权限的 IAM 角色，请参阅 [在 Amazon OpenSearch Ingestion 中设置角色和用户](pipeline-security-overview.md)。 #### 步骤 1：配置管道角色 完成管道先决条件设置后，[配置管道角色](pipeline-security-overview.md#pipeline-security-sink)以在管道配置中使用，并为该角色添加以下权限： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "SecretsManagerReadAccess", "Effect": "Allow", "Action": [ "secretsmanager:GetSecretValue" ], "Resource": ["arn:aws:secretsmanager:us-east-1:111122223333:secret:secret-name"] }, { "Effect": "Allow", "Action": [ "ec2:AttachNetworkInterface", "ec2:CreateNetworkInterface", "ec2:CreateNetworkInterfacePermission", "ec2:DeleteNetworkInterface", "ec2:DeleteNetworkInterfacePermission", "ec2:DetachNetworkInterface", "ec2:DescribeNetworkInterfaces" ], "Resource": [ "arn:aws:ec2:*:*:network-interface/*", "arn:aws:ec2:*:*:subnet/*", "arn:aws:ec2:*:*: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 摄取管道的 IAM 角色提供上述 Amazon EC2 权限，因为管道使用这些权限在您的 VPC 中创建和删除网络接口。该管道只能通过此网络接口访问 Kafka 集群。 #### 步骤 2：创建管道 然后，您可以配置如下所示的 OpenSearch Ingestion 管道，将 Kafka 指定为来源。 您可以将多个 OpenSearch 服务域指定为数据的目的地。此功能允许有条件地路由或将传入数据复制到多个 OpenSearch 服务域。 您还可以将数据从源 Confluent Kafka 集群迁移到无服务器 V OpenSearch PC 集合。务必要在管道配置中提供网络访问策略。您可以使用 Confluent 架构注册表来定义 Confluent 架构。 ``` version: "2" kafka-pipeline: source: kafka: encryption: type: "ssl" topics: - name: "topic-name" group_id: "group-id" bootstrap_servers: - "bootstrap-server.us-east-1.aws.private.confluent.cloud:9092" authentication: sasl: plain: username: ${aws_secrets:confluent-kafka-secret:username} password: ${aws_secrets:confluent-kafka-secret:password} schema: type: confluent registry_url: https://my-registry.us-east-1.aws.confluent.cloud api_key: "${{aws_secrets:schema-secret:schema_registry_api_key}}" api_secret: "${{aws_secrets:schema-secret:schema_registry_api_secret}}" basic_auth_credentials_source: "USER_INFO" sink: - opensearch: hosts: ["https://search-mydomain.us-east-1.es.amazonaws.com"] aws: region: "us-east-1" index: "confluent-index" extension: aws: secrets: confluent-kafka-secret: secret_id: "my-kafka-secret" region: "us-east-1" schema-secret: secret_id: "my-self-managed-kafka-schema" region: "us-east-1" ``` 您可以使用预先配置的蓝图，以创建此管道。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 # 使用 Amazon OpenSearch Ingestion 从自行管理的 OpenSearch 集群迁移数据 您可以使用带有自我管理 OpenSearch 或 Elastic OpenSearch search 的 Amazon Ingestion 管道将数据迁移到亚马逊 OpenSearch 服务域和无服务器集合。 OpenSearch OpenSearch Ingestion 支持公用和私有网络配置，用于迁移来自自管理 OpenSearch 和 Elasticsearch 的数据。 ## 从公共 OpenSearch 集群迁移 您可以使用 OpenSearch Ingestion 管道从具有公共配置的自管理集群 OpenSearch 或 Elasticsearch 集群迁移数据，这意味着可以公开解析域 DNS 名称。为此，请设置一个以自我管理 OpenSearch 或 Elast OpenSearch icsearch 为源，以 OpenSearch 服务或 OpenSearch 无服务器为目标的采集管道。这可以有效地将您的数据从自我管理的源集群迁移到托管的 AWS目标域或集合。 ### 先决条件 在创建 OpenSearch 摄取管道之前，请执行以下步骤： 1. 创建包含您要迁移的数据的自管理集群 OpenSearch 或 Elastisearch 集群，并配置公有 DNS 名称。有关说明，请参阅 OpenSearch 文档中的[创建集群](https://opensearch.org/docs/latest/tuning-your-cluster/)。 1. 创建要将数据迁移到的 OpenSearch 服务域或 OpenSearch 无服务器集合。有关更多信息，请参阅[创建 OpenSearch 服务域](createupdatedomains.md#createdomains)和[创建集合](serverless-create.md)。 1. 使用在您的自管理集群上设置身份验证。 AWS Secrets Manager按照 [Rotate AWS Secrets Manager secrets](https://docs.aws.amazon.com/secretsmanager/latest/userguide/rotating-secrets.html) 中的步骤启用密钥轮换。 1. 将[基于资源的策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/ac.html#ac-types-resource)附加到域，或将[数据访问策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/serverless-data-access.html)附加到集合。这些访问策略允许 OpenSearch Ingestion 将数据从您的自管理集群写入您的域或集合。 以下示例域访问策略允许您在下一步中创建的管道角色将数据写入域。确保使用自身 ARN 更新 `resource`。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "AWS": "arn:aws:iam::444455556666:role/pipeline-role" }, "Action": [ "es:DescribeDomain", "es:ESHttp*" ], "Resource": [ "arn:aws:es:us-east-1:111122223333:domain/domain-name" ] } ] } ``` ------ 要创建具有向该集合或域写入数据的正确权限的 IAM 角色，请参阅 [在 Amazon OpenSearch Ingestion 中设置角色和用户](pipeline-security-overview.md)。 ### 步骤 1：配置管道角色 设置好 OpenSearch 管道先决条件后，[配置要在管道配置中使用的管道角色](pipeline-security-overview.md#pipeline-security-sink)，添加写入 OpenSearch 服务域或 OpenSearch 无服务器集合的权限，以及从 Secrets Manager 读取密钥的权限。 ### 步骤 2：创建管道 然后，您可以配置如下所示的 OpenSearch Ingestion 管道，该管道指定 OpenSearch 为来源。 您可以指定多个 OpenSearch 服务域作为数据的目的地。此功能允许有条件地路由或将传入数据复制到多个 OpenSearch 服务域。 您还可以将数据从源集群 OpenSearch 或 Elasticsearch 集群迁移到 OpenSearch 无服务器 VPC 集合。务必要在管道配置中提供网络访问策略。 ``` version: "2" opensearch-migration-pipeline: source: opensearch: acknowledgments: true host: [ "https://my-self-managed-cluster-name:9200" ] indices: include: - index_name_regex: "include-.*" exclude: - index_name_regex: '\..*' authentication: username: ${aws_secrets:secret:username} password: ${aws_secrets:secret:password} scheduling: interval: "PT2H" index_read_count: 3 start_time: "2023-06-02T22:01:30.00Z" sink: - opensearch: hosts: ["https://search-mydomain.us-east-1.es.amazonaws.com"] aws: region: "us-east-1" #Uncomment the following lines if your destination is an OpenSearch Serverless collection #serverless: true # serverless_options: # network_policy_name: "network-policy-name" index: "${getMetadata(\"opensearch-index\")}" document_id: "${getMetadata(\"opensearch-document_id\")}" enable_request_compression: true dlq: s3: bucket: "bucket-name" key_path_prefix: "apache-log-pipeline/logs/dlq" region: "us-east-1" extension: aws: secrets: secret: secret_id: "my-opensearch-secret" region: "us-east-1" refresh_interval: PT1H ``` 您可以使用预先配置的蓝图来创建此管道。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 ## 从 VPC 中的 OpenSearch 集群迁移数据 您还可以使用 OpenSearch 摄取管道从 VPC 中运行的自管理集群 OpenSearch 或 Elasticsearch 集群迁移数据。为此，请设置一个以自我管理 OpenSearch 或 Elast OpenSearch icsearch 为源，以 OpenSearch 服务或 OpenSearch 无服务器为目标的采集管道。这可以有效地将您的数据从自我管理的源集群迁移到托管的 AWS目标域或集合。 ### 先决条件 在创建 OpenSearch 摄取管道之前，请执行以下步骤： 1. 使用包含要迁移的数据的 VPC 网络配置创建自管理集群 OpenSearch 或 Elastisearch 集群。 1. 创建要将数据迁移到的 OpenSearch 服务域或 OpenSearch 无服务器集合。有关更多信息，请参阅[创建 OpenSearch 服务域](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/createupdatedomains.html#createdomains)和[创建集合](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/serverless-manage.html#serverless-create)。 1. 使用在您的自管理集群上设置身份验证。 AWS Secrets Manager按照 [Rotate AWS Secrets Manager secrets](https://docs.aws.amazon.com/secretsmanager/latest/userguide/rotating-secrets.html) 中的步骤启用密钥轮换。 1. 获取有权访问自管 OpenSearch 或弹性搜索的 VPC 的 ID。选择 OpenSearch Ingestion 要使用的 VPC CIDR。 **注意** 如果您使用创建管道，则还必须将您的 OpenSearch 摄取管道连接到您的 VPC 才能使用自我管理 OpenSearch 或 Elasticsearch。 AWS 管理控制台 要执行此操作，请找到**源网络选项**部分，选择**连接到 VPC** 复选框，然后从提供的任意一个默认选项中选择 CIDR。您可以按照 [RFC 1918 当前最佳实践](https://datatracker.ietf.org/doc/html/rfc1918)中的定义，使用私有地址空间中的任何 CIDR。 要提供自定义 CIDR，请从下拉菜单中选择**其他**。为避免接入和自行管理之间的 IP 地址冲突 OpenSearch，请确保自管理 VP OpenSearch C CIDR 与用于 OpenSearch 摄取的 CIDR 不同。 OpenSearch 1. 将[基于资源的策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/ac.html#ac-types-resource)附加到域，或将[数据访问策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/serverless-data-access.html)附加到集合。这些访问策略允许 OpenSearch Ingestion 将数据从您的自管理集群写入您的域或集合。 以下示例域访问策略允许您在下一步中创建的管道角色将数据写入域。确保使用自身 ARN 更新 `resource`。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "AWS": "arn:aws:iam::444455556666:role/pipeline-role" }, "Action": [ "es:DescribeDomain", "es:ESHttp*" ], "Resource": [ "arn:aws:es:us-east-1:111122223333:domain/example.com" ] } ] } ``` ------ 要创建具有向该集合或域写入数据的正确权限的 IAM 角色，请参阅 [在 Amazon OpenSearch Ingestion 中设置角色和用户](pipeline-security-overview.md)。 ### 步骤 1：配置管道角色 完成管道先决条件设置后，[配置管道角色](pipeline-security-overview.md#pipeline-security-sink)以在管道配置中使用，并为该角色添加以下权限： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "SecretsManagerReadAccess", "Effect": "Allow", "Action": [ "secretsmanager:GetSecretValue" ], "Resource": ["arn:aws:secretsmanager:us-east-1:111122223333:secret:secret-name"] }, { "Effect": "Allow", "Action": [ "ec2:AttachNetworkInterface", "ec2:CreateNetworkInterface", "ec2:CreateNetworkInterfacePermission", "ec2:DeleteNetworkInterface", "ec2:DeleteNetworkInterfacePermission", "ec2:DetachNetworkInterface", "ec2:DescribeNetworkInterfaces" ], "Resource": [ "arn:aws:ec2:*:*:network-interface/*", "arn:aws:ec2:*:*:subnet/*", "arn:aws:ec2:*:*: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 摄取管道的 IAM 角色提供上述 Amazon EC2 权限，因为管道使用这些权限在您的 VPC 中创建和删除网络接口。管道只能通过此网络接口访问 OpenSearch 集群。 ### 步骤 2：创建管道 然后，您可以配置如下所示的 OpenSearch Ingestion 管道，该管道指定 OpenSearch 为来源。 您可以指定多个 OpenSearch 服务域作为数据的目的地。此功能允许有条件地路由或将传入数据复制到多个 OpenSearch 服务域。 您还可以将数据从源集群 OpenSearch 或 Elasticsearch 集群迁移到 OpenSearch 无服务器 VPC 集合。务必要在管道配置中提供网络访问策略。 ``` version: "2" opensearch-migration-pipeline: source: opensearch: acknowledgments: true host: [ "https://my-self-managed-cluster-name:9200" ] indices: include: - index_name_regex: "include-.*" exclude: - index_name_regex: '\..*' authentication: username: ${aws_secrets:secret:username} password: ${aws_secrets:secret:password} scheduling: interval: "PT2H" index_read_count: 3 start_time: "2023-06-02T22:01:30.00Z" sink: - opensearch: hosts: ["https://search-mydomain.us-east-1.es.amazonaws.com"] aws: region: "us-east-1" #Uncomment the following lines if your destination is an OpenSearch Serverless collection #serverless: true # serverless_options: # network_policy_name: "network-policy-name" index: "${getMetadata(\"opensearch-index\")}" document_id: "${getMetadata(\"opensearch-document_id\")}" enable_request_compression: true dlq: s3: bucket: "bucket-name" key_path_prefix: "apache-log-pipeline/logs/dlq" region: "us-east-1" extension: aws: secrets: secret: secret_id: "my-opensearch-secret" region: "us-east-1" refresh_interval: PT1H ``` 您可以使用预先配置的蓝图来创建此管道。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 # 使用采 OpenSearch 集管道和 Amazon Kinesis Data Streams 使用 OpenSearch 采集管道和 Amazon Kinesis Data Streams，将来自多个流的流记录数据提取到亚马逊 OpenSearch 服务域和集合中。 OpenSearch Ingestion 管道整合了流媒体摄取基础架构，为持续从 Kinesis 摄取直播记录提供了一种高规模、低延迟的方式。 **Topics** + [Amazon Kinesis Data Streams 作为源](#confluent-cloud-kinesis) + [Amazon Kinesis Data Streams 跨账户作为源](#kinesis-cross-account-source) ## Amazon Kinesis Data Streams 作为源 通过以下过程，您将学习如何设置使用 Amazon K OpenSearch inesis Data Streams 作为数据源的摄取管道。本节介绍必要的先决条件，例如创建 OpenSearch 服务域或 OpenSearch 无服务器集合，并逐步完成配置管道角色和创建管道的步骤。 ### 先决条件 要设置管道，则需要一个或多个活跃的 Kinesis Data Streams。这些流必须正在接收记录，或准备接收来自其他源的记录。有关更多信息，请参阅[ OpenSearch 摄取概述](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/osis-getting-started-tutorials.html)。 **设置管道** 1. **创建 OpenSearch 服务域或 OpenSearch 无服务器集合** 要创建域名或集合，请参阅 [ OpenSearch Ingestion 入门](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/osis-getting-started-tutorials.html)。 要创建具有向该集合或域写入数据的正确权限的 IAM 角色，请参阅 [Resource-based policies](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/ac.html#ac-types-resource)。 1. **配置具有权限的管道角色** 设置要在管道配置中使用的[管道角色](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/pipeline-security-overview.html#pipeline-security-sink)，并为其添加以下权限。将 *placeholder values* 替换为您自己的信息。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "allowReadFromStream", "Effect": "Allow", "Action": [ "kinesis:DescribeStream", "kinesis:DescribeStreamConsumer", "kinesis:DescribeStreamSummary", "kinesis:GetRecords", "kinesis:GetShardIterator", "kinesis:ListShards", "kinesis:ListStreams", "kinesis:ListStreamConsumers", "kinesis:RegisterStreamConsumer", "kinesis:SubscribeToShard" ], "Resource": [ "arn:aws:kinesis:us-east-1:111122223333:stream/stream-name" ] } ] } ``` ------ 如果已在流上启用服务器端加密，则以下 AWS KMS 策略允许对记录进行解密。将 *placeholder values* 替换为您自己的信息。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "allowDecryptionOfCustomManagedKey", "Effect": "Allow", "Action": [ "kms:Decrypt", "kms:GenerateDataKey" ], "Resource": "arn:aws:kms:us-east-1:111122223333:key/key-id" } ] } ``` ------ 为使管道能够将数据写入域，域必须具有[域级访问策略](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/ac.html#ac-types-resource)，以允许 **sts\$1role\$1arn** 管道角色访问域。 以下示例是域访问策略，允许在上一步（`pipeline-role`）中创建的管道角色将数据写入 `ingestion-domain` 域。将 *placeholder values* 替换为您自己的信息。 ``` { "Statement": [ { "Effect": "Allow", "Principal": { "AWS": "arn:aws:iam::your-account-id:role/pipeline-role" }, "Action": ["es:DescribeDomain", "es:ESHttp*"], "Resource": "arn:aws:es:AWS 区域:account-id:domain/domain-name/*" } ] } ``` 1. **创建管道** 配置一个将 **K** 指定inesis-data-streams为源的 OpenSearch 摄取管道。您可以在 OpenSearch Ingestion 控制台上找到用于创建此类管道的现成蓝图。（可选）要使用创建管道 AWS CLI，您可以使用名为 “**`AWS-KinesisDataStreamsPipeline`**” 的蓝图。将 *placeholder values* 替换为您自己的信息。 ``` version: "2" kinesis-pipeline: source: kinesis_data_streams: acknowledgments: true codec: # Based on whether kinesis records are aggregated or not, you could choose json, newline or ndjson codec for processing the records. # JSON codec supports parsing nested CloudWatch Events into individual log entries that will be written as documents into OpenSearch. # json: # key_name: "logEvents" # These keys contain the metadata sent by CloudWatch Subscription Filters # in addition to the individual log events: # include_keys: [ 'owner', 'logGroup', 'logStream' ] newline: streams: - stream_name: "stream name" # Enable this if ingestion should start from the start of the stream. # initial_position: "EARLIEST" # checkpoint_interval: "PT5M" # Compression will always be gzip for CloudWatch, but will vary for other sources: # compression: "gzip" - stream_name: "stream name" # Enable this if ingestion should start from the start of the stream. # initial_position: "EARLIEST" # checkpoint_interval: "PT5M" # Compression will always be gzip for CloudWatch, but will vary for other sources: # compression: "gzip" # buffer_timeout: "1s" # records_to_accumulate: 100 # Change the consumer strategy to "polling". Default consumer strategy will use enhanced "fan-out" supported by KDS. # consumer_strategy: "polling" # if consumer strategy is set to "polling", enable the polling config below. # polling: # max_polling_records: 100 # idle_time_between_reads: "250ms" aws: # Provide the Role ARN with access to Amazon Kinesis Data Streams. This role should have a trust relationship with osis-pipelines.amazonaws.com sts_role_arn: "arn:aws:iam::111122223333:role/Example-Role" # Provide the AWS 区域 of the Data Stream. region: "us-east-1" sink: - opensearch: # Provide an Amazon OpenSearch Serverless domain endpoint hosts: [ "https://search-mydomain-1a2a3a4a5a6a7a8a9a0a9a8a7a.us-east-1.es.amazonaws.com" ] index: "index_${getMetadata(\"stream_name\")}" # Ensure adding unique document id as a combination of the metadata attributes available. document_id: "${getMetadata(\"partition_key\")}_${getMetadata(\"sequence_number\")}_${getMetadata(\"sub_sequence_number\")}" aws: # Provide a Role ARN with access to the domain. This role should have a trust relationship with osis-pipelines.amazonaws.com sts_role_arn: "arn:aws:iam::111122223333:role/Example-Role" # Provide the AWS 区域 of the domain. region: "us-east-1" # Enable the 'serverless' flag if the sink is an Amazon OpenSearch Serverless collection serverless: false # serverless_options: # Specify a name here to create or update network policy for the serverless collection # network_policy_name: "network-policy-name" # Enable the 'distribution_version' setting if the OpenSearch Serverless domain is of version Elasticsearch 6.x # distribution_version: "es6" # Enable and switch the 'enable_request_compression' flag if the default compression setting is changed in the domain. See https://docs.aws.amazon.com/opensearch-service/latest/developerguide/gzip.html # enable_request_compression: true/false # Optional: Enable the S3 DLQ to capture any failed requests in an S3 bucket. Delete this entire block if you don't want a DLQ. dlq: s3: # Provide an S3 bucket bucket: "your-dlq-bucket-name" # Provide a key path prefix for the failed requests # key_path_prefix: "kinesis-pipeline/logs/dlq" # Provide the region of the bucket. region: "us-east-1" # Provide a Role ARN with access to the bucket. This role should have a trust relationship with osis-pipelines.amazonaws.com sts_role_arn: "arn:aws:iam::111122223333:role/Example-Role" ``` **配置选项** 有关 Kinesis 配置选项，请参阅[文档中的*OpenSearch*配置选项](https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/sources/kinesis/#configuration-options)。 **可用的元数据属性** + **stream\$1name**：从中摄取记录的 Kinesis Data Streams 的名称 + **partition\$1key**：正在摄取的 Kinesis Data Streams 记录的分区键 + **sequence\$1number**：正在摄取的 Kinesis Data Streams 记录的序列号 + **sub\$1sequence\$1number**：正在摄取的 Kinesis Data Streams 记录的子序列号 1. **（可选）为 Kinesis Data Streams 管道配置推荐的计算单位 (OCUs)** 也可以将 OpenSearch Kinesis Data Streams 源管道配置为从多个流中提取直播记录。为加快数据摄取速度，我们建议您为每个新增流添加一个额外的计算单位。 ### 数据一致性 OpenSearch Ingestion 支持 end-to-end确认以确保数据的持久性。当管道从 Kinesis 读取流记录时，会根据与流关联的分片动态分配读取流记录的工作。Pipeline 在收录 OpenSearch 域或集合中的所有记录后收到确认后，将自动对流进行检查点。这将避免对流记录进行重复处理。 要根据流名称创建索引，请在 opensearch 接收器部分中将索引定义为 **"index\$1\$1\$1getMetadata(\$1"stream\$1name\$1")\$1"**。 ## Amazon Kinesis Data Streams 跨账户作为源 您可以通过 Amazon Kinesis Data Streams 跨账户授予访问权限 OpenSearch ，以便摄取管道可以访问另一个账户作为来源的 Kinesis Data Streams。请完成以下步骤以启用跨账户访问。 **配置跨账户访问** 1. **在拥有 Kinesis 流的账户中设置资源策略** 将 *placeholder values* 替换为您自己的信息。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "StreamReadStatementID", "Effect": "Allow", "Principal": { "AWS": "arn:aws:iam::111122223333:role/Pipeline-Role" }, "Action": [ "kinesis:DescribeStreamSummary", "kinesis:GetRecords", "kinesis:GetShardIterator", "kinesis:ListShards" ], "Resource": "arn:aws:kinesis:us-east-1:444455556666:stream/stream-name" }, { "Sid": "StreamEFOReadStatementID", "Effect": "Allow", "Principal": { "AWS": "arn:aws:iam::111122223333:role/Pipeline-Role" }, "Action": [ "kinesis:DescribeStreamSummary", "kinesis:ListShards" ], "Resource": "arn:aws:kinesis:us-east-1:444455556666:stream/stream-name/consumer/consumer-name" } ] } ``` ------ 1. **（可选）设置消费端和消费端资源策略** 此步骤为可选操作，仅当您计划使用增强型扇出消费端策略读取流记录时才需要执行。有关更多信息，请参阅[开发具有专用吞吐量增强型扇出消费端](https://docs.aws.amazon.com/streams/latest/dev/enhanced-consumers.html)。 1. **设置消费端** 要重复使用现有消费端，可跳过此步骤。有关更多信息，请参阅[RegisterStreamConsumer](https://docs.aws.amazon.com/dms/latest/APIReference/API_RegisterStreamConsumer.html)《*亚马逊 Kinesis Data Streams API* 参考》。 在以下示例 CLI 命令中，*placeholder values*用您自己的信息替换。 **Example 示例 CLI 命令：** ``` aws kinesis register-stream-consumer \ --stream-arn "arn:aws:kinesis:AWS 区域:account-id:stream/stream-name" \ --consumer-name consumer-name ``` 1. **设置消费端资源策略** 在以下语句中，*placeholder values*用您自己的信息替换。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "ConsumerEFOReadStatementID", "Effect": "Allow", "Principal": { "AWS": "arn:aws:iam::111122223333:role/Pipeline-Role" }, "Action": [ "kinesis:DescribeStreamConsumer", "kinesis:SubscribeToShard" ], "Resource": "arn:aws:kinesis:us-east-1:444455556666:stream/stream-1/consumer/consumer-name" } ] } ``` ------ 1. **管道配置** 对于跨账户数据摄取，请为每个流的 `kinesis_data_streams` 下添加以下属性： + `stream_arn`：该流所属账户中存在的流的 ARN + `consumer_arn`：这是可选属性，如果选择默认的增强型扇出消费端策略，则必须指定该属性。指定此字段的实际消费端 ARN。将 *placeholder values* 替换为您自己的信息。 ``` version: "2" kinesis-pipeline: source: kinesis_data_streams: acknowledgments: true codec: newline: streams: - stream_arn: "arn:aws:kinesis:region:stream-account-id:stream/stream-name" consumer_arn: "consumer arn" # Enable this if ingestion should start from the start of the stream. # initial_position: "EARLIEST" # checkpoint_interval: "PT5M" - stream_arn: "arn:aws:kinesis:region:stream-account-id:stream/stream-name" consumer_arn: "consumer arn" # initial_position: "EARLIEST" # buffer_timeout: "1s" # records_to_accumulate: 100 # Enable the consumer strategy to "polling". Default consumer strategy will use enhanced "fan-out" supported by KDS. # consumer_strategy: "polling" # if consumer strategy is set to "polling", enable the polling config below. # polling: # max_polling_records: 100 # idle_time_between_reads: "250ms" aws: # Provide the Role ARN with access to Kinesis. This role should have a trust relationship with osis-pipelines.amazonaws.com sts_role_arn: "arn:aws:iam::111122223333:role/Example-Role" # Provide the AWS 区域 of the domain. region: "us-east-1" sink: - opensearch: # Provide an OpenSearch Serverless domain endpoint hosts: [ "https://search-mydomain-1a2a3a4a5a6a7a8a9a0a9a8a7a.us-east-1.es.amazonaws.com" ] index: "index_${getMetadata(\"stream_name\")}" # Mapping for documentid based on partition key, shard sequence number and subsequence number metadata attributes document_id: "${getMetadata(\"partition_key\")}_${getMetadata(\"sequence_number\")}_${getMetadata(\"sub_sequence_number\")}" aws: # Provide a Role ARN with access to the domain. This role should have a trust relationship with osis-pipelines.amazonaws.com sts_role_arn: "arn:aws:iam::111122223333:role/Example-Role" # Provide the AWS 区域 of the domain. region: "us-east-1" # Enable the 'serverless' flag if the sink is an OpenSearch Serverless collection serverless: false # serverless_options: # Specify a name here to create or update network policy for the serverless collection # network_policy_name: network-policy-name # Enable the 'distribution_version' setting if the OpenSearch Serverless domain is of version Elasticsearch 6.x # distribution_version: "es6" # Enable and switch the 'enable_request_compression' flag if the default compression setting is changed in the domain. See https://docs.aws.amazon.com/opensearch-service/latest/developerguide/gzip.html # enable_request_compression: true/false # Optional: Enable the S3 DLQ to capture any failed requests in an S3 bucket. Delete this entire block if you don't want a DLQ. dlq: s3: # Provide an Amazon S3 bucket bucket: "your-dlq-bucket-name" # Provide a key path prefix for the failed requests # key_path_prefix: "alb-access-log-pipeline/logs/dlq" # Provide the AWS 区域 of the bucket. region: "us-east-1" # Provide a Role ARN with access to the bucket. This role should have a trust relationship with osis-pipelines.amazonaws.com sts_role_arn: "arn:aws:iam::111122223333:role/Example-Role" ``` 1. **OSI 管道角色 Kinesis Data Streams** 1. **IAM 策略** 将以下策略附加到管道角色。将 *placeholder values* 替换为您自己的信息。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "kinesis:DescribeStreamConsumer", "kinesis:SubscribeToShard" ], "Resource": [ "arn:aws:kinesis:us-east-1:111122223333:stream/my-stream" ] }, { "Sid": "allowReadFromStream", "Effect": "Allow", "Action": [ "kinesis:DescribeStream", "kinesis:DescribeStreamSummary", "kinesis:GetRecords", "kinesis:GetShardIterator", "kinesis:ListShards", "kinesis:ListStreams", "kinesis:ListStreamConsumers", "kinesis:RegisterStreamConsumer" ], "Resource": [ "arn:aws:kinesis:us-east-1:111122223333:stream/my-stream" ] } ] } ``` ------ 1. **信任策略** 要从流账户摄取数据，您需要在管道摄取角色与流账户之间建立信任关系。将以下内容添加到管道角色。将 *placeholder values* 替换为您自己的信息。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [{ "Effect": "Allow", "Principal": { "AWS": "arn:aws:iam::111122223333:root" }, "Action": "sts:AssumeRole" }] } ``` ------ ## 后续步骤 将数据导出到管道后，您可以从配置为管道接收器的 OpenSearch 服务域中[查询](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/searching.html)数据。以下资源可帮助您开始使用： + [Amazon OpenSearch 服务中的可观察性](observability.md) + [发现痕迹](observability-analyze-traces.md) + [Amazon OpenSearch 服务中的可观察性](observability.md) # 将 OpenSearch 摄取管道与 AWS Lambda 使用[AWS Lambda 处理](https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/processors/aws-lambda/)器使用自定义代码丰富来自 OpenSearch Ingestion 支持的任何源或目标的数据。使用 Lambda 处理器，您可以应用自己的数据转换或丰富，然后将处理后的事件返回到您的管道，以供进一步处理。该处理器支持自定义数据处理，让您能够在数据进入处理流程前完全掌控其操作方式。 **注意** Lambda 处理器处理单个事件的有效载荷大小限制为 5 MB。此外，Lambda 处理器仅支持 JSON 数组格式的响应。 ## 先决条件 使用 Lambda 处理器创建管道之前，创建以下资源： + 一种 AWS Lambda 用于丰富和转换源数据的函数。有关说明，请参阅[创建首个 Lambda 函数](https://docs.aws.amazon.com/lambda/latest/dg/getting-started.html)。 + 将成为管道接收器的 OpenSearch 服务域或 OpenSearch 无服务器集合。有关更多信息，请参阅[创建 OpenSearch 服务域](createupdatedomains.md#createdomains)和[创建集合](serverless-create.md)。 + 包含向域或集合接收器写入权限的管道角色。有关更多信息，请参阅 [管道角色](pipeline-security-overview.md#pipeline-security-sink)。 管道角色还需附加权限策略，以允许其调用管道配置中指定的 Lambda 函数。例如： ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "allowinvokeFunction", "Effect": "Allow", "Action": [ "lambda:invokeFunction", "lambda:InvokeAsync", "lambda:ListFunctions" ], "Resource": "arn:aws:lambda:us-east-1:111122223333:function:function-name" } ] } ``` ------ ## 创建管道 要 AWS Lambda 用作处理器，请配置 OpenSearch Ingestion 管道并指定`aws_lambda`为处理器。您也可以使用 **AWS Lambda 自定义增强**蓝图创建管道。有关更多信息，请参阅 [使用蓝图](pipeline-blueprint.md)。 以下示例管道从 HTTP 源接收数据，使用日期处理器和处理器对其进行丰富，然后将 AWS Lambda 处理后的数据提取到域中。 OpenSearch ``` version: "2" lambda-processor-pipeline: source: http: path: "/${pipelineName}/logs" processor: - date: destination: "@timestamp" from_time_received: true - aws_lambda: function_name: "my-lambda-function" tags_on_failure: ["lambda_failure"] batch: key_name: "events" aws: region: region sink: - opensearch: hosts: [ "https://search-mydomain.us-east-1es.amazonaws.com" ] index: "table-index" aws: region: "region" serverless: false ``` 以下示例 AWS Lambda 函数通过向提供的事件数组中的每个元素添加新的键值对 (`"transformed": "true"`) 来转换传入的数据，然后发回修改后的版本。 ``` import json def lambda_handler(event, context): input_array = event.get('events', []) output = [] for input in input_array: input["transformed"] = "true"; output.append(input) return output ``` ## 批处理 管道向 Lambda 处理器发送批处理事件，并动态调整批量大小，以确保其始终低于 5 MB 的限制。 下图是管道批处理的示例： ``` batch: key_name: "events" input_arrary = event.get('events', []) ``` **注意** 创建管道时，确保 Lambda 处理器配置中的 `key_name` 选项与 Lambda 处理程序中的事件密钥相匹配。 ## 条件筛选 条件筛选允许您根据事件数据中的特定条件控制 AWS Lambda 处理器何时调用 Lambda 函数。当您希望有选择地处理某些类型的事件，同时忽略其他事件时，这尤其有用。 以下示例配置使用条件筛选： ``` processors: - aws_lambda: function_name: "my-lambda-function" aws: region: "region" lambda_when: "/sourceIp == 10.10.10.10" ```