本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

RDS for PostgreSQL

完成以下步骤，使用适用于 Post OpenSearch greSQL 的 Amazon RDS for RDS 配置摄取管道。

适用于 PostgreSQL 的 RDS 先决条件

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

步骤 1：配置管道角色

设置 Amazon RDS 管道先决条件后，配置要在管道配置中使用的管道角色。此外，还要为该角色添加 Amazon RDS 源的以下权限：

{
  "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": "allowNetworkInterfacesGroup1",
      "Effect": "Allow",
      "Action": [
        "ec2:CreateNetworkInterface",
        "ec2:CreateNetworkInterfacePermission"
      ],
      "Resource": [
        "arn:aws:ec2:*:account-id:network-interface/*",
        "arn:aws:ec2:*:account-id:subnet/*",
        "arn:aws:ec2:*:account-id:security-group/*"
      ]
    },
    {
      "Sid": "allowNetworkInterfacesGroup2",
      "Effect": "Allow",
      "Action": [
        "ec2:AttachNetworkInterface",
        "ec2:CreateNetworkInterface",
        "ec2:CreateNetworkInterfacePermission",
        "ec2:DeleteNetworkInterface",
        "ec2:DeleteNetworkInterfacePermission",
        "ec2:DetachNetworkInterface",
        "ec2:DescribeNetworkInterfaces"
      ],
      "Resource": [
        "arn:aws:ec2:*:account-id:network-interface/*",
        "arn:aws:ec2:*:account-id:subnet/*",
        "arn:aws:ec2:*:account-id:security-group/*"
      ],
      "Condition": {
        "StringEquals": {
          "aws:RequestTag/OSISManaged": "true"
        }
      }
    },
    {
      "Sid": "allowDescribeEC2",
      "Effect": "Allow",
      "Action": [
        "ec2:Describe*"
      ],
      "Resource": "*"
    },
    {
      "Sid": "allowTagCreation",
      "Effect": "Allow",
      "Action": [
        "ec2:CreateTags"
      ],
      "Resource": "arn:aws:ec2:*:account-id:network-interface/*",
      "Condition": {
        "StringEquals": {
          "aws:RequestTag/OSISManaged": "true"
        }
      }
    },
    {
      "Sid": "AllowDescribeInstances",
      "Effect": "Allow",
      "Action": [
        "rds:DescribeDBInstances"
      ],
      "Resource": [
        "arn:aws:rds:region:account-id:db:*"
      ]
    },
    {
      "Sid": "AllowDescribeClusters",
      "Effect": "Allow",
      "Action": [
        "rds:DescribeDBClusters"
      ],
      "Resource": [
        "arn:aws:rds:region:account-id:cluster:DB-id"
      ]
    },
    {
      "Sid": "AllowSnapshots",
      "Effect": "Allow",
      "Action": [
        "rds:DescribeDBClusterSnapshots",
        "rds:CreateDBClusterSnapshot",
        "rds:AddTagsToResource"
      ],
      "Resource": [
        "arn:aws:rds:region:account-id:cluster:DB-id",
        "arn:aws:rds:region:account-id:cluster-snapshot:DB-id*"
      ]
    },
    {
      "Sid": "AllowExport",
      "Effect": "Allow",
      "Action": [
        "rds:StartExportTask"
      ],
      "Resource": [
        "arn:aws:rds:region:account-id:cluster:DB-id",
        "arn:aws:rds:region:account-id:cluster-snapshot:DB-id*"
      ]
    },
    {
      "Sid": "AllowDescribeExports",
      "Effect": "Allow",
      "Action": [
        "rds:DescribeExportTasks"
      ],
      "Resource": "*",
      "Condition": {
        "StringEquals": {
          "aws:RequestedRegion": "region",
          "aws:ResourceAccount": "account-id"
        }
      }
    },
    {
      "Sid": "AllowAccessToKmsForExport",
      "Effect": "Allow",
      "Action": [
        "kms:Decrypt",
        "kms:Encrypt",
        "kms:DescribeKey",
        "kms:RetireGrant",
        "kms:CreateGrant",
        "kms:ReEncrypt*",
        "kms:GenerateDataKey*"
      ],
      "Resource": [
        "arn:aws:kms:region:account-id:key/export-key-id"
      ],
    },
    {
      "Sid": "AllowPassingExportRole",
      "Effect": "Allow",
      "Action": "iam:PassRole",
      "Resource": [
        "arn:aws:iam::account-id:role/export-role"
      ]
    },
    {
      "Sid": "SecretsManagerReadAccess",
      "Effect": "Allow",
      "Action": [
        "secretsmanager:GetSecretValue"
      ],
      "Resource": [
        "arn:aws:secretsmanager:*:account-id: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 Aurora 作为来源，您需要为管道配置 VPC 访问权限。您选择的 VPC 应与您的 Amazon Aurora 来源使用的 VPC 相同。然后选择一个或多个子网以及一个或多个 VPC 安全组。请注意，管道需要对 Aurora MySQL 数据库进行网络访问，因此您还应验证您的 Aurora 集群是否配置了 VPC 安全组，该安全组允许从管道的 VPC 安全组到数据库端口的入站流量。有关更多信息，请参阅使用安全组控制访问权限。

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

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

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

数据一致性

该管道通过持续轮询或接收来自 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 自动确定字段类型。您还可以 OpenSearch 通过映射模板明确定义最适合您的字段类型。

下表列出了适用于 PostgreSQL 的 RDS 数据类型和相应的字段类型。 OpenSearch OpenSearch 如果未定义显式映射，则 “默认 OpenSearch 字段类型” 列将显示相应的字段类型。在这种情况下，使用动态映射 OpenSearch 自动确定字段类型。“推荐 OpenSearch 字段类型” 列是在映射模板中明确指定的相应推荐字段类型。这些字段类型与 RDS for PostgreSQL 中的数据类型更加一致，通常可以启用中提供的更好的搜索功能。 OpenSearch

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

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

限制

在为 RDS for PostgreSQL 设置 OpenSearch 摄取管道时，请考虑以下限制：

推荐 CloudWatch 警报

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