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: パイプラインを作成する

そして、ソースとして RDS for PostgreSQL インスタンスを指定する OpenSearch Ingestion パイプラインを次のように設定できます。

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 である必要があります。次に、1 つ以上のサブネットと 1 つ以上の VPC セキュリティグループを選択します。パイプラインには Aurora MySQL データベースへのネットワークアクセスが必要であるため、パイプラインの VPC セキュリティグループからデータベースポートへのインバウンドトラフィックを許可する VPC セキュリティグループが Aurora クラスターに設定されていることも確認する必要があります。詳細については、「セキュリティグループによるアクセス制御」を参照してください。

AWS マネジメントコンソール を使用してパイプラインを作成する場合は、Amazon Aurora をソースとして使用するために VPC にパイプラインをアタッチする必要もあります。これを行うには、[ネットワーク設定] セクションを見つけ、[VPC にアタッチ] を選択し、提供されたデフォルトオプションのいずれかから CIDR を選択するか、独自の CIDR を選択します。RFC 1918 のベストカレントプラクティスで定義されているように、プライベートアドレス空間から任意の CIDR を使用できます。

カスタム CIDR を指定するには、ドロップダウンメニューから [その他] を選択します。OpenSearch Ingestion と Amazon RDS 間の IP アドレスの衝突を回避するには、Amazon Aurora VPC の CIDR が OpenSearch Ingestion の CIDR と異なることを確認してください。

詳細については、「Configuring VPC access for a pipeline」を参照してください。

データ整合性

パイプラインは、Amazon RDS インスタンスから変更を継続的にポーリングまたは受信し、OpenSearch インデックス内の対応するドキュメントを更新することで、データ整合性を確保します。

OpenSearch Ingestion は、データの耐久性を確保するためにエンドツーエンドの確認応答をサポートしています。パイプラインがスナップショットまたはストリームを読み取る際に、並列処理用のパーティションが動的に作成されます。パイプラインは、OpenSearch ドメインまたはコレクション内のすべてのレコードを取り込んだ後に確認応答を受信すると、パーティションを完了としてマークします。OpenSearch Serverless 検索コレクションに取り込みたい場合は、パイプラインでドキュメント ID を生成できます。OpenSearch Serverless 時系列コレクションに取り込みたい場合は、パイプラインがドキュメント 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 メトリクスをお勧めします。これらのメトリクスは、エクスポートから処理されたデータ量、ストリームから処理されたイベント数、エクスポートとストリームイベントの処理エラー、宛先に書き込まれたドキュメントの数を特定するのに役立ちます。これらのメトリクスの 1 つが、指定された時間にわたって指定された値を超えた場合にアクションを実行するように CloudWatch アラームを設定できます。