本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。 # 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总计 | 该指标指示该流中处理失败的记录总数。 |