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

RDS for PostgreSQL

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

RDS for PostgreSQL 先决条件

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

步骤 1：配置管道角色

完成 Amazon RDS 管道先决条件设置后，配置管道角色，以便在管道配置中使用。还需为该角色添加以下针对 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 摄取管道，该管道将 RDS for PostgreSQL 实例指定为源。

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

要使用适用于 PostgreSQL 的 RDS 作为源，您需要为管道配置 VPC 访问权限。您选择的 VPC 应与您的 Amazon RDS 源所使用的 VPC 相同。然后选择一个或多个子网以及一个或多个 VPC 安全组。请注意，管道需要通过网络访问适用于 PostgreSQL 的 RDS 数据库，因此您还应验证您的 Amazon RDS 实例是否配置了 VPC 安全组，该安全组允许从管道的 VPC 安全组到数据库端口的入站流量。有关更多信息，请参阅使用安全组控制访问权限。

如果您使用创建管道，则还必须将管道连接到 VPC，才能使用适用于 PostgreSQL 的 RDS 作为源。 AWS 管理控制台 要执行此操作，请找到网络配置部分，选择连接到 VPC，然后从提供的任意一个默认选项中选择 CIDR，或者选择自己的 CIDR。CIDR 块必须使用 /24 前缀长度。您可以使用 RFC 1918 最佳现行实践中定义的私有地址空间中的任何 /24 CIDR。

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

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

数据一致性

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

OpenSearch 摄取支持端到端确认，以确保数据的耐久性。管道读取快照或流时，它会动态创建分区以进行并行处理。当管道在载入 OpenSearch 域或集合中的所有记录后收到确认时，该管道将分区标记为已完成。如果您想采集到 OpenSearch 无服务器搜索集合中，则可以在管道中生成文档 ID。如果您想采集到 OpenSearch Serverless 时间序列集合中，请注意，该管道不会生成文档 ID，因此您必须在流水线接收器配置document_id: "${getMetadata(\"primary_key\")}"中省略。

OpenSearch 摄取管道还将传入的事件操作映射到相应的批量索引操作中，以帮助采集文档。这样可以保持数据的一致性，这样 Amazon RDS 中的每项数据更改都会与中 OpenSearch相应的文档更改保持一致。

映射数据类型

OpenSearch 摄取管道将 PostgreSQL 数据类型映射到适合 OpenSearch 服务域或集合使用的表示形式。如果中未定义映射模板 OpenSearch，则根据第一个发送的文档使用动态映射 OpenSearch 自动确定字段类型。您还可以在映射模板中 OpenSearch 明确定义最适合您的字段类型。

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

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

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

限制

为适用于 PostgreSQL 的 RDS 设置 OpenSearch 摄取管道时，请考虑以下限制：

推荐 CloudWatch 警报

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