DynamoDBとOpenSearch Serviceのzero-ETL統合で何ができるようになるのかを試す

はじめに

先日、AWS re:Invent2023が開催されました。今回のアップデートの多くが生成AIに関連しており、これまでのre:Inventとは一線を画すような内容でした。

なかでも、Amazon CTOのWerner VogelsがKeynoteにて「You don't have good data, you don't have good AI」と述べたように、単に生成AI周りのサービスだけではなく、生成AIの精度の大きなファクターとなるデータの利活用にも注力をしている印象を持ちました。

写真はhttps://www.youtube.com/watch?v=UTRBVPvzt9w より引用

さて、今回の発表の中でアップデートの一つとして、Amazon DynamoDBのAmazon OpenSearch Serviceとのzero-ETL統合の発表がありました。

この統合により、DynamoDBのデータを強力な検索で活用できるようになり、生成AIでのRAG活用も期待されます。
私自身、DynamoDBを使う機会が多く、検索がもっと簡単にできるようならないかなと思っていたので、嬉しいアップデートです！
本記事ではこの機能の解説と簡単な使用方法、今後の発展について考えてみたいと思います。

zero-ETLとは

zero-ETLとは公式ドキュメント上で下記のように定義されています。

ゼロ ETL は、ETL データパイプラインを構築する必要性を排除し、または最小限に抑える一連の統合です。抽出、変換、ロード (ETL) は、さまざまなソースからのデータを結合、クリーニング、正規化して、分析、人工知能 (AI)、機械学習 (ML) のワークロードに対応できるようにするプロセスです。

従来のデータ分析では、データを別のソースに移動させるために複雑なETL（Extract, Transform, Load）パイプラインの構築が必要でした。しかし、このようなパイプラインの構築と運用を不要にする新しいアプローチが登場しています。それが「zero-ETL」と呼ばれるデータ統合の仕組みです。

zero-ETLとは、データを統合することで従来のETLプロセスを不要としたアプローチです。これにより、データ移動や変換の複雑さが減少し、データの即時利用が可能になります。企業はETLの構築や運用の負担なしに、必要な場所にデータを配置し、強力な検索機能を用いてデータを活用できるようになります。結果として、組織はデータを最大限活用できるようになります。

AWSはこのzero-ETLへの取り組みを以前から進めてきましたが、今回紹介する機能以外にも複数のAWSデータベースサービスがRedshiftへのzero-ETL統合に対応するアップデートが発表されるなど、AWSがzero-ETLにより力を入れていることが伺えました。

写真はhttps://www.youtube.com/watch?v=PMfn9_nTDbM&t=3772s より引用

Amazon OpenSearch Serviceとは

まず初めに、OpenSearchとは、ElasticSearchからフォークしたオープンソースのソフトウェアです。このソフトウェアは、データを集約し、柔軟な検索および分析機能を提供することで知られています。

AWSが提供するAmazon OpenSearch Serviceは、OpenSearchをAWS上で容易にデプロイ、運用、スケーリングできるようにしたマネージドサービスです。このサービスによって、ユーザーはOpenSearchの強力な機能をAWSのクラウド環境で手軽に利用できます。

また従来、ユーザーはノードやクラスタの管理を行う必要がありましたが、これらの管理が不要なAmazon OpenSearch Serverlessも提供されるようになりました。このサーバレスタイプは、管理の手間を削減し、スケーリングと運用の効率を高めることで、より簡単かつ効率的にデータ分析を行うことが可能になっています。

DynamoDBとOpenSearch Serviceのzero-ETL統合とは

今回のアップデートにより、DynamoDBとAmazon OpenSearch Service間でのETL統合が一般公開となりました。これにより、ETLの構築や運用の必要性をほぼなくし、DynamoDBのデータをAmazon OpenSearch Serviceに複製するプロセスを簡素化します。これにより、OpenSearchの強力な検索機能をDynamoDBのデータに対して利用できるようになります。

特に注目すべきは、この統合によって、全文検索、あいまい検索、オートコンプリートなどの柔軟な検索オプションがDynamoDBに保存されたデータで実現可能になる点です。さらに、ベクトル検索を用いてレコメンデーションエンジン（RAG）を構築し、生成AIアプリケーション上でDynamoDBのデータに応じた幅広い回答を提供することも可能になります。

個人的な観点から見ると、DynamoDBはそのスケーリング能力に優れていますが、検索性については複雑な検索ができず、検索したい特定の属性に対してグローバルセカンダリインデックスを個別に設定する必要などもありました。
そのため、今回のアップデートはDynamoDB上のデータの利活用性をさらに高めたものだと思います。

DynamoDBとAmazon OpenSearch Serviceのzero-ETL統合の構成・仕組み

DynamoDBとAmazon OpenSearch Serviceをzero-ETLで統合する構成は下記のようになります。

図を見て分かる通り、zero-ETL統合はOpenSearch Ingestionが担います。
OpenSearch Ingestionはフルマネージド型のサーバーレスデータコレクターで、OpenSearch Serviceドメインと OpenSearch Serverlessにリアルタイムログ、メトリクス、トレースデータを配信することができます。

OpenSearch IngestionがDynamoDBからOpenSearchにデータを送信するためには、大きく２つの手順を実施します。
まず１つ目が赤色の線で示したポイント・イン・タイム・リカバリ（PITR）を使用したスナップショットの取り込みです。このステップでは、DynamoDBのデータを特定の時点でスナップショットとして取得し、それをS3に保存します。その後、保存されたスナップショットをOpenSearchに取り込むことで、特定時点のデータの完全なコピーを同期します。

２つ目が青色の線で示したDynamoDB Streamsを使用した継続的な変更データキャプチャの取り込みです。一度特定時点のデータを同期した後、その後の差分データはDynamoDB Streamsを通じて継続的にキャプチャし、Amazon OpenSearch Serviceに取り込まれます。

これらのステップにより、Amazon OpenSearch Serviceにほぼリアルタイムでかつ完全なデータを取り込むことが可能になります。

DynamoDBのzero-ETL統合を試してみる

事前準備

zero-ETLを構築する前の事前準備として下記の対応が必要です。

1. バケットの作成
   Amazon OpenSearch Serviceに取り込むためのPITRスナップショット保存用のS3バケットを作成しておく必要があります。

2. パイプライン用のロールの作成
   公式ドキュメントに記載のOpenSearch Ingestionにアタッチする用のIAMロールを作成します。

3. DynamoDBテーブルの作成
   zero-ETLを試すDynamoDBのテーブルを予め作成しておく必要があります。
   今回の記事においては、Usersというテーブルを作成し、userIdをパーティションキーとして設定しています。
   なお、サンプルデータはfakerというライブラリを活用して、作成しました。

OpenSearch Serverlessの構築

はじめにOpenSearch Serverlessの構築を行います。
AWSコンソール上でAmazon Opensearch serviceのページを開き、左側のメニューからコレクションを選択します。

コレクションの作成ページが表示されますので、「コレクションの作成」ボタンを選択します。

ここでt登場したコレクションとはAmazon OpenSearch Serviceにおけるクラスタ（ドメイン）に相当する論理的なインデックスの集合のことを指します。

ここでは任意のコレクション名を入力します。
次にコレクション名の下にコレクションタイプを選択する項目があります。このコレクションタイプとはデータの検索種類ごとに用意されており、それぞれのタイプごとにキャッシュ戦略や利用できるAPIが異なります。

違いについては、Blackbeltの資料の下記の図がとてもわかり易いです。

今回はお試しなので「検索」を選択することとします。

他の設定はデフォルトのまま次に進みます。

確認画面が表示されるので、「送信」を選択します。

実行してからしばらくするとコレクションの作成が完了し、下記のようにエンドポイントURLが発行されます。

この情報は後ほどのETL統合の設定やクエリ検索時に使用します。

DynamoDBとAmazon OpenSearch Service間のzero-ETL統合の設定

次にDynamoDBとAmazon OpenSearch Service間のzero-ETL統合のためにOpenSearch Ingestionの設定を行います。
DynamoDBのコンソールにアクセスし、左のメニューから「インテグレーション」を選択します。

インテグレーションの管理ページが表示され、ページ上部に「Turn on point-in-time-recovery(PITR) and DynamoDB stream」と記載されたバナーが表示されます。

DynamoDB zero-ETLの構成の章で説明した通り、DynamoDBとAmazon OpenSearch Serviceの統合を行うためにはPITRとDynamoDB Streamの設定が必要となりますので、バナー内の右側にある「オンにする」ボタンを選択します。

「パイプラインを作成」ボタンが活性になりますので、選択します。

次にパイプラインの作成画面が表示されます。パイプライン名には任意の名前を入力します。

またページ下部の「パイプラインの構成」という項目には、DynamoDBからAmazon OpenSearch Serviceにデータを取り込むためのパイプライン設定をYAML形式で記述する必要があります。

なお、パイプラインの各ユースケーステンプレートYAMLが用意されており、そのテンプレートをもとに定義を作成することができます。
テンプレートを使用するためには、「パイプラインの構成」の横にある「設定のブループリント」というドロップダウンメニューのボタンを選択します。すると、メニューに今回のユースケース用のテンプレートである「AWS-DynamoDBChangeDataCapturePipeline」が表示されるので、これを選択します。

DynamoDB用のテンプレートYAML定義が下記のように出力されます。

このテンプレートを参考に私は下記のように定義しました。

version: "2"
dynamodb-pipeline:
  source:
    dynamodb:
      acknowledgments: true
      tables:
        - table_arn: "arn:aws:dynamodb:us-east-1:XXXXXXX:table/Users"
          stream:
            start_position: "LATEST"
          export:
            # zero-ETLのためのデータの中間保存用のバケット
            s3_bucket: "ddb-opensearch-bucket-xxxxx"
            s3_region: "us-east-1"
            s3_prefix: "ddb-to-opensearch-export/"
      aws:
      　　　　# 先に作成したパイプライン用のロールのARN
        sts_role_arn: "arn:aws:iam::XXXXXXX:role/OpenSearchIngestionPipelineRole"
        region: "us-east-1"
  sink:
    - opensearch:
 　　　　　　　　　　　　　　# OpenSearchのコレクションのエンドポイントURLを設定   
        hosts: [ "https://xxxxxxxxxx.us-east-1.aoss.amazonaws.com" ]
        index: "users-index"
        action: "${getMetadata(\"opensearch_action\")}"
        aws:
          # 先に作成したパイプライン用のロールのARN
          sts_role_arn: "arn:aws:iam::XXXXXXX:role/OpenSearchIngestionPipelineRole"
          region: "us-east-1"
          serverless: true

上記のYAML定義において、S3の情報とsts_role_arnには事前準備の章で作成したS3とIAM RoleのARNを入力します。またOpenSearchのhostsには、先に作成したOpenSearch ServerlessのコレクションのエンドポイントURLを設定します。
上記のように編集後、「保存」を選択します。

設定後、暫くしてOpensearch Serverlessのコンソール上のインデックスタブを開くと、インデックスが作成されデータが転送されていることが確認できます。

それではこの転送されたインデックス内のデータの検索が行えるのか確認をしてみます。

転送されたデータを検索してみる

今回、データの検索にはGoogle Colabを使用しました。

まずはAWS SDKを実行するために必要なIAMユーザのアクセスキーとシークレットアクセスキーを定義します。

import getpass
AWS_ACCESS_KEY_ID=input("AWS_ACCESS_KEY_ID:")
AWS_SECRET_ACCESS_KEY=getpass.getpass("AWS_SECRET_ACCESS_KEY:")

次に上記のキー情報を使用して、認証した上で、Opensearchのクライアントを初期化します。
ここでｈostにはOpenSearch Collectionのエンドポイント情報を設定する必要があります。

from boto3 import Session
from opensearchpy import OpenSearch, RequestsAWSV4SignerAuth, RequestsHttpConnection

session = Session(aws_access_key_id=AWS_ACCESS_KEY_ID, aws_secret_access_key=AWS_SECRET_ACCESS_KEY)
credentials = session.get_credentials()

host = 'xxxxxx.us-east-1.aoss.amazonaws.com'
port = 443

auth = RequestsAWSV4SignerAuth(credentials, "us-east-1", "aoss")

client = OpenSearch(
        hosts=[{"host": host, "port": 443}],
        http_auth=auth,
        use_ssl=True,
        verify_certs=True,
        connection_class=RequestsHttpConnection,
        timeout=30,
)

これで準備は完了です。
クエリを投げてユーザデータを検索してみたいと思います。

まず初めにユーザ情報のcompanyが特定の名称で完全一致するユーザ情報を取得してみたいと思います。
検索するためのクエリはJSON形式で下記のように定義できます。

query = {
    "query": {
      "match": {
        "company": "Garrett LLC",
    }
  }
}

下記のOpenSearchクラスのメソッドの引数に上記の変数queryを渡すことで検索が実行できます。

client.search(query, index="users-index")

結果は下記の通り、問題なく一致するユーザ情報を取得することができました。

{'took': 90,
 'timed_out': False,
 '_shards': {'total': 0, 'successful': 0, 'skipped': 0, 'failed': 0},
 'hits': {'total': {'value': 1, 'relation': 'eq'},
  'max_score': 1.3862942,
  'hits': [{'_index': 'users-index',
    '_id': 'dnIxM4wBVEkIstROR0ir',
    '_score': 1.3862942,
    '_source': {'userId': 'c63e3575-4213-4a01-9805-7d10055d92cd',
     'company': 'Garrett LLC',
     'phone_number': '203-866-8129x53060',
     'text': 'Word family part thought this first cultural. Still direction notice sister. Walk special cause south open life agreement.\nParticipant behavior capital per price daughter everything.',
     'address': '7352 Yu Light Suite 311\nRichardsonside, CA 70257',
     'email': '[e-mail]',
     'name': 'Richard Lewis',
     'job': 'Merchant navy officer'}}]}}

次にcompanyにGarretもしくはGreenを含むユーザ情報を取得するクエリを作成して検索してみます。

query =   {
  "query": {
    "match": {
      "company": {
        "query": "Garrett Green",
        "operator": "or",
      }
    }
  }
}
client.search(query, index="users-index")

結果は下記のように一致する２件のユーザ情報を取得することができました。

{'took': 36,
 'timed_out': False,
 '_shards': {'total': 0, 'successful': 0, 'skipped': 0, 'failed': 0},
 'hits': {'total': {'value': 2, 'relation': 'eq'},
  'max_score': 0.6931471,
  'hits': [{'_index': 'users-index',
    '_id': 'dnIxM4wBVEkIstROR0ir',
    '_score': 0.6931471,
    '_source': {'userId': 'c63e3575-4213-4a01-9805-7d10055d92cd',
     'company': 'Garrett LLC',
     'phone_number': '203-866-8129x53060',
     'text': 'Word family part thought this first cultural. Still direction notice sister. Walk special cause south open life agreement.\nParticipant behavior capital per price daughter everything.',
     'address': '7352 Yu Light Suite 311\nRichardsonside, CA 70257',
     'email': '[e-mail]',
     'name': 'Richard Lewis',
     'job': 'Merchant navy officer'}},
   {'_index': 'users-index',
    '_id': 'zLMyM4wBeLRb-CEDYMIe',
    '_score': 0.6931471,
    '_source': {'userId': '0a85db81-1629-452f-9898-37d120765956',
     'company': 'Green-Booker',
     'phone_number': '413.871.4451',
     'text': 'Floor represent require station soldier sense. Our onto recognize yeah leader against.',
     'address': '60000 Wright Falls Apt. 766\nMorrisonhaven, MD 82113',
     'email': '[e-mail]',
     'name': 'Willie Reid',
     'job': 'Multimedia programmer'}}]}}

上記はあくまで一例で、様々な検索条件で情報を取得することが可能です。
詳しくは下記のOpensearchのドキュメントを参照ください。

試したけどできなかったこと

実は上記以外にもいろいろ試してみたのですが、上手く行かなかったものがあるので紹介します。

1. DynamoDBデータのベクトル検索

今回はベクトル検索を紹介したかったのですが、上手く行かず掲載することができませんでした...OpenSearch内のベクトル検索用のコレクションへのデータ取り込みまでは上手く行ったものの、ベクトル検索を試しても何もヒットしないというあと一歩届かない結果になってしまいました。
ただ下記のセッションではベクトル検索の設定方法についても言及しているので、おそらく可能だと思います（セッション内で言われていることは一通り試したのですが、私の環境では上手く行きませんでした）。

現状、セッションで言及されているほどの詳細な内容がドキュメントには書かれていないので、ドキュメントのアップデートを待ちたいと思います。

2. Knowledge base for Amazon Bedrockと組み合わせたRAG構成

また今回アップデートでGAとなったKnowledge base for Amazon Bedrockを活用したRAG構成も検証してみました。

しかし、現在Knowledge baseのデータソースは現状S3のみしかサポートされていなようで、ベクトル検索版のAmazon OpenSearch Serviceを統合することはできたものの、S3からのデータがSyncされてしまい、DynamoDBのデータをRAGとして扱うことはできませんでした（そもそもベクトル検索ができなかったので、どのみち上手く行かなかったかもしれません）。
今後のアップデートでKnowlegde baseがS3以外のデータソースにも対応することに期待したいと思います。

今回の検証では、最も私が実現したかったDynamoDBデータのRAG活用には一歩届きませんでしたが、リトライして再度ブログを書きたいと思います。

最後に

この機能によって、DynamoDBからAmazon OpenSearch Serviceに簡単にETLが実現できるようになり、DynamoDBのデータをAmazon OpenSearch Serviceにコピーして柔軟な検索ができるようになりました。
試してできなかったことにも書いた通り、今回はDynamoDBのデータを活用したベクトル検索はご紹介できませんでしたが、DynamoDBのデータをRAGに活用することによってLLMによってできることが拡充していくことは非常に楽しみだと思わせるアップデートでした。

試した場合はリソースの削除は忘れないようにしてください。
特にAmazon OpenSearch Serviceはかなり高額なサービスなので、注意してください！