RDS for PostgreSQL

完成以下步骤，使用 Amazon RDS for PostgreSQL 配置 OpenSearch Ingestion 管道。

RDS for PostgreSQL 先决条件

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

步骤 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 Ingestion 管道，将 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

要使用 Amazon Aurora 作为源，您需要为管道配置 VPC 访问权限。您选择的 VPC 应与 Amazon Aurora 源使用的 VPC 相同。然后选择一个或多个子网以及一个或多个 VPC 安全组。请注意，该管道需要访问 Aurora MySQL 数据库的网络权限，因此您还需确认 Aurora 集群已配置 VPC 安全组，允许来自管道 VPC 安全组的入站流量访问数据库端口。有关更多信息，请参阅使用安全组控制访问权限。

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

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

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

数据一致性

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

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

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

映射数据类型

OpenSearch Ingestion 管道将 PostgreSQL 数据类型映射为适合 OpenSearch Service 域或集合消费的表示形式。如果 OpenSearch 中未定义映射模板，OpenSearch 将根据首次发送的文档自动通过动态映射确定字段类型。您还可通过映射模板，在 OpenSearch 中显式定义最适合您的字段类型。

下表列出 RDS for PostgreSQL 数据类型及相应的 OpenSearch 字段类型。如果未定义显式映射，则默认 OpenSearch 字段类型列会显示 OpenSearch 中相应的字段类型。在此情况下，OpenSearch 会通过动态映射自动确定字段类型。推荐的 OpenSearch 字段类型列是在映射模板中显式指定的相应推荐字段类型。这些字段类型与 RDS for PostgreSQL 中的数据类型更加一致，通常能更好地发挥 OpenSearch 提供的搜索功能。

我们建议在 OpenSearch Ingestion 管道中配置死信队列（DLQ）。如果已配置该队列，OpenSearch Service 会将所有因动态映射失败而无法摄取的失败文档发送到该队列。

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

限制

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

推荐 CloudWatch 警报

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