DynamoDBとOpenSearchサーバーレスのZero-ETLを設定してみた

DynamoDBのデータをOpenSearchサーバーレスにコード無しでつなぎこめるZero-ETLが発表されて気になっていたので実際に試してみた手順を紹介します
統計値を求めたい声が社内で多く上がり、毎回エンジニアがPartiQLとピボットテーブルでいい感じに統計を出し続けるのにも限界を感じて試してみました
実験で設定したのでInfrastruce as Codeにはなってないです

DynamoDBテーブルの作成

複数あってもOKです
PITRとStreamの有効化が必須ですがそれ以外はなんでも大丈夫そうです

S3バケットの作成

PITRのデータを一時的に保管するのに必要なんだとか
デフォルトの設定のままでよいので一つ作成します

OpenSearch サーバーレスコレクションの作成

インデックスをテーブルごとに分ければ一つのコレクションで済みます
統計値を求めるならひとつの方が都合がいいかもしれません
統計値を求めるためなのでコレクションタイプは検索を選択し、内部でしか使用しないのでアクティブレプリカは無効化しました
セキュリティもとりあえず簡単作成を利用

Dashboard（Kibanaですね）にアクセスするため、コレクション作成時に一緒に作成されたネットワークポリシーを編集し、パブリックなOpenSearch Dashboardsへのアクセスを有効にします

IAMロールの作成

ETLを実行するOpenSearch Ingestion用のIAMロールを作成します
作成のテンプレートで「OpenSearch Ingestion Pipelines」を選択し、なんのポリシーもアタッチせずに作成します

そしてインラインポリシーを以下の内容で作成します

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "allowRunExportJob",
      "Effect": "Allow",
      "Action": [
        "dynamodb:DescribeTable",
        "dynamodb:DescribeContinuousBackups",
        "dynamodb:ExportTableToPointInTime"
      ],
      "Resource": ["arn:aws:dynamodb:{region}:{accountId}:table/{tableName}"]
    },
    {
      "Sid": "allowCheckExportjob",
      "Effect": "Allow",
      "Action": ["dynamodb:DescribeExport"],
      "Resource": [
        "arn:aws:dynamodb:{region}:{accountId}:table/{tableName}/export/*"
      ]
    },
    {
      "Sid": "allowReadFromStream",
      "Effect": "Allow",
      "Action": [
        "dynamodb:DescribeStream",
        "dynamodb:GetRecords",
        "dynamodb:GetShardIterator"
      ],
      "Resource": [
        "arn:aws:dynamodb:{region}:{accountId}:table/{tableName}/stream/*"
      ]
    },
    {
      "Sid": "allowReadAndWriteToS3ForExport",
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:AbortMultipartUpload",
        "s3:PutObject",
        "s3:PutObjectAcl"
      ],
      "Resource": ["arn:aws:s3:::{bucketName}/*"]
    },
    {
      "Sid": "allowReadToOpenSearchCollection",
      "Effect": "Allow",
      "Action": ["aoss:BatchGetCollection", "aoss:APIAccessAll"],
      "Resource": [
        "arn:aws:aoss:{region}:{accountId}:collection/{collectionId}"
      ]
    },
    {
      "Sid": "allowReadToOpenSearchSecurityPolicy",
      "Effect": "Allow",
      "Action": [
        "aoss:CreateSecurityPolicy",
        "aoss:GetSecurityPolicy",
        "aoss:UpdateSecurityPolicy"
      ],
      "Resource": "*",
      "Condition": {
        "StringEquals": {
          "aoss:collection": "{collectionName}"
        }
      }
    }
  ]
}

DynamoDBのテーブルを複数取り込みたい場合は配列になっているので複数指定するだけでOKです
このポリシーでDynamoDBへアクセスし、S3にデータをバックアップし、OpenSearchにアクセスできるようにしています

IAMロールの信頼

IAMロールのARNをOpenSearchサーバーレスコレクションのデータアクセスに設定されている easy-{コレクション名} というポリシーの編集画面で、一つだけ設定されているルールのプリンシパルに追加します

OpenSearch 取り込みパイプラインの作成

DynamoDBテーブルを複数用意した場合このパイプラインはテーブルの数だけ必要になります
「パイプラインを作成」をクリックするとブループリントの選択を求められますがここはBlankでOKです
「パイプラインの構成」には以下のyamlを設定します

version: "2"
dynamodb-pipeline:
  source:
    dynamodb:
      acknowledgments: true
      tables:
        - table_arn: "arn:aws:dynamodb:{region}:{accountId}:table/{tableName}"
          stream:
            start_position: "LATEST"
          export:
            s3_bucket: "{bucketName}"
            s3_region: "{region}"
            s3_prefix: "ddb-to-opensearch-export/{tableName}"
      aws:
         sts_role_arn: "{IAMロールのARN}"
         region: "{region}"
  sink:
    - opensearch:
        hosts: [ "{OpenSearch エンドポイント}" ]
        index: "{インデックス名}"
        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: "{IAMロールのARN}"
          region: "{region}"
          serverless: true
          serverless_options:
            network_policy_name: "easy-{コレクション名}"

ネットワークはパブリックアクセスにします
あとはデフォルトのままで作成に進みます
作成したパイプラインの詳細画面からCloudWatch ロググループを開きテーリングをすると進捗が分かりやすくお勧めです

OpenSearch Dashboardsで統計値を求める

マネコンにリンクがあります
左カラムメニューからVisualizeを選ぶとindex patternを作るよう求められるので設定したindexを含むパターンを作成します
いくつかデータが入っていればTime fieldを選べるはずなので何か設定しておくとVisualizeしやすかったりします

苦労したこと

IAMロールのポリシーが公式の例だと足りず、とはいえ無駄には付与したくなくて何度かトライ&エラーすることになりました
あとDashboardがサーバーレスのコールドスタートなのかアクセスを許可しているのに401が出て驚きましたね