Amazon DynamoDB OpenSearch での取り込みパイプラインの使用

DynamoDB で Ingestion OpenSearch パイプラインを使用して、DynamoDB テーブルイベント (作成、更新、削除など) を Amazon OpenSearch Service ドメインとコレクションにストリーミングできます。 OpenSearch Ingestion パイプラインには、変更データキャプチャ (CDC) インフラストラクチャが組み込まれており、DynamoDB テーブルからデータを継続的にストリーミングするための、高スケールで低レイテンシーの手段を提供します。

データを処理するソースとして DynamoDB を使用するには、完全な初期スナップショットを使用する方法と使用しない方法の 2 つがあります。

完全な初期スナップショットは、DynamoDB がpoint-in-time リカバリ (PITR) 機能を使用して作成するテーブルのバックアップです。DynamoDB はこのスナップショットを Amazon S3 にアップロードします。そこから、取り込みパイプラインはドメイン内の 1 OpenSearch つのインデックスに送信するか、ドメイン内の複数のインデックスにパーティション化します。DynamoDB 内のデータを OpenSearch 一貫性に保つために、パイプラインは DynamoDB テーブル内のすべての作成、更新、削除イベントを OpenSearch 、インデックスまたはインデックスに保存されたドキュメントと同期します。

完全な初期スナップショットを使用すると、 OpenSearch 取り込みパイプラインは最初にスナップショットを取り込んでから、DynamoDB Streams からのデータの読み取りを開始します。最終的には、DynamoDB と の間のほぼリアルタイムのデータ整合性が追いついて維持されます OpenSearch。このオプションを選択するときは、テーブルで PITRと DynamoDB ストリームの両方を有効にする必要があります。

DynamoDB OpenSearch との Ingestion 統合を使用して、スナップショットなしでイベントをストリーミングすることもできます。他のメカニズムからの完全なスナップショットが既にある場合、または DynamoDB Streams を使用して DynamoDB テーブルから現在のイベントをストリーミングするだけの場合は、このオプションを選択します。このオプションを選択した場合、必要なのは、テーブルで DynamoDB ストリームを有効にすることだけです。

この統合の詳細については、「 Amazon DynamoDB デベロッパーガイド」の「Amazon OpenSearch Service との ETLDynamoDB ゼロ統合」を参照してください。

前提条件

パイプラインを設定するには、DynamoDB Streams が有効になっている DynamoDB テーブルが必要です。ストリームでは NEW_IMAGE ストリームビュータイプを使用する必要があります。ただし、このストリームビュータイプがユースケースに適合するNEW_AND_OLD_IMAGES場合、 OpenSearch 取り込みパイプラインは でイベントをストリーミングすることもできます。

スナップショットを使用している場合は、テーブルで point-in-time 復旧を有効にする必要もあります。詳細については、「Amazon DynamoDB デベロッパーガイド」の「テーブルの作成」、 point-in-time 「復旧の有効化」、「ストリームの有効化」を参照してください。

ステップ 1: パイプラインロールを設定する

DynamoDB テーブルを設定したら、パイプライン設定で使用するパイプラインロールを設定し、そのロールに次の DynamoDB 許可を追加します。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "allowRunExportJob",
            "Effect": "Allow",
            "Action": [
                "dynamodb:DescribeTable",
                "dynamodb:DescribeContinuousBackups",
                "dynamodb:ExportTableToPointInTime"
            ],
            "Resource": [
                "arn:aws:dynamodb:us-east-1:{account-id}:table/my-table"
            ]
        },
        {
            "Sid": "allowCheckExportjob",
            "Effect": "Allow",
            "Action": [
                "dynamodb:DescribeExport"
            ],
            "Resource": [
                "arn:aws:dynamodb:us-east-1:{account-id}:table/my-table/export/*"
            ]
        },
        {
            "Sid": "allowReadFromStream",
            "Effect": "Allow",
            "Action": [
                "dynamodb:DescribeStream",
                "dynamodb:GetRecords",
                "dynamodb:GetShardIterator"
            ],
            "Resource": [
                "arn:aws:dynamodb:us-east-1:{account-id}:table/my-table/stream/*"
            ]
        },
        {
            "Sid": "allowReadAndWriteToS3ForExport",
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:AbortMultipartUpload",
                "s3:PutObject",
                "s3:PutObjectAcl"
            ],
            "Resource": [
                "arn:aws:s3:::my-bucket/{exportPath}/*"
            ]
        }
    ]
}

AWS KMS カスタマーマネージドキーを使用して、エクスポートデータファイルを暗号化することもできます。エクスポートされたオブジェクトを復号するには、パイプラインのエクスポート設定において、次の形式でキー ID として s3_sse_kms_key_id を指定します: arn:aws:kms:us-west-2:{account-id}:key/my-key-id。次のポリシーには、カスタマーマネージドキーを使用するために必要なアクセス許可が含まれています。

{
    "Sid": "allowUseOfCustomManagedKey",
    "Effect": "Allow",
    "Action": [
        "kms:GenerateDataKey",
        "kms:Decrypt"
    ],
    "Resource": arn:aws:kms:us-west-2:{account-id}:key/my-key-id
}

ステップ 2: パイプラインを作成する

その後、DynamoDB OpenSearch をソースとして指定する Ingestion パイプラインを次のように設定できます。このサンプルパイプラインは、PITRスナップショットtable-aを使用して からデータを取り、その後に DynamoDB Streams からのイベントを取り込みます。開始位置が LATEST であることは、パイプラインが DynamoDB Streams から最新のデータを読み取る必要があることを示します。

version: "2"
cdc-pipeline:
  source:
    dynamodb:
      tables:
      - table_arn: "arn:aws:dynamodb:us-west-2:{account-id}:table/table-a"  
        export:
          s3_bucket: "my-bucket"
          s3_prefix: "export/"
        stream:
          start_position: "LATEST"
      aws:
        region: "us-west-2"
        sts_role_arn: "arn:aws:iam::{account-id}:role/pipeline-role"
  sink:
  - opensearch:
      hosts: ["https://search-mydomain.us-east-1.es.amazonaws.com"]
      index: "${getMetadata(\"table_name\")}"
      index_type: custom
      normalize_index: true
      document_id: "${getMetadata(\"primary_key\")}"
      action: "${getMetadata(\"opensearch_action\")}"
      document_version: "${getMetadata(\"document_version\")}"
      document_version_type: "external"

事前設定された DynamoDB ブループリントを使用して、このパイプラインを作成できます。詳細については、「ブループリントを使用したパイプラインの作成」を参照してください。

データ整合性

OpenSearch 取り込みでは、データの耐久性を確保するため end-to-end の承認がサポートされています。パイプラインがスナップショットまたはストリームを読み取る際に、並列処理用のパーティションが動的に作成されます。パイプラインは、 OpenSearch ドメインまたはコレクション内のすべてのレコードを取り込んだ後に確認を受け取ると、パーティションを完了としてマークします。

OpenSearch Serverless 検索コレクションに取り込む場合は、パイプラインでドキュメント ID を生成できます。 OpenSearch Serverless 時系列コレクションに取り込む場合は、パイプラインがドキュメント ID を生成しないことに注意してください。

Ingestion OpenSearch パイプラインは、受信イベントアクションを対応する一括インデックス作成アクションにマッピングして、ドキュメントの取り込みにも役立ちます。これによりデータの整合性が保たれるため、DynamoDB のすべてのデータ変更が、 の対応するドキュメントの変更と照合されます OpenSearch。

データ型のマッピング

OpenSearch サービスは、各受信ドキュメントのデータ型を DynamoDB の対応するデータ型に動的にマッピングします。次の表は、 OpenSearch サービスがさまざまなデータ型を自動的にマッピングする方法を示しています。

{
 "template": {
  "mappings": {
   "properties": {
    "MixedNumberAttribute": {
     "type": "float"
    }
   }
  }
 }
}

{
 "template": {
  "mappings": {
   "properties": {
    "PartType": {
     "type": "keyword"
    }
   }
  }
 }
}

{
 "template": {
  "mappings": {
   "properties": {
    "ImageData": {
     "type": "binary"
    }
   }
  }
 }
}

{
 "template": {
  "mappings": {
   "properties": {
    "AdditionalDescriptions": {
     "properties": {
      "PartType": {
       "type": "keyword"
      }
     }
    }
   }
  }
 }
}

Ingestion パイプラインでデッドレターキュー (DLQ) OpenSearch を設定することをお勧めします。キューを設定した場合、 OpenSearch 動的マッピングの失敗により取り込むことができないすべての失敗したドキュメントがサービスからキューに送信されます。

自動マッピングが失敗した場合は、パイプライン設定で template_type と template_content を使用して明示的なマッピングルールを定義できます。あるいは、パイプラインを開始する前に、検索ドメインまたはコレクションにマッピングテンプレートを直接作成することもできます。

制限事項

DynamoDB OpenSearch の取り込みパイプラインを設定するときは、次の制限事項を考慮してください。

DynamoDB の推奨 CloudWatch アラーム

取り込みパイプラインのパフォーマンスをモニタリングするには、次の CloudWatch メトリクスをお勧めします。これらのメトリクスは、エクスポートから処理されたデータの量、ストリームから処理されたイベントの量、エクスポートとストリームイベントの処理エラー、および宛先に書き込まれたドキュメントの数を特定するのに役立ちます。これらのメトリクスの 1 つが指定された時間、指定された値を超えたときにアクションを実行するように CloudWatch アラームを設定できます。