はじめに こんにちは。WEARバックエンド部SREブロックの 春日 です。普段は WEAR というサービスのSREとして開発・運用に携わっています。 本記事では、WEARのハイブリッド検索のリリースに伴い刷新した検索インデクシングシステム（以下、インデクサー）について、 OpenSearch Ingestion を採用しようとした際にハマったポイントや、ベクトル検索のためのインデクサーを設計する上で工夫した点を中心に紹介します。 目次 はじめに 目次 背景 既存のインデクサーと刷新の動機 ベクトルデータの保持方法の検討 インデクサーの構成方針 BigQuery → S3 のデータ連携 日次更新の設計 差分更新の設計 初期設計：OpenSearch Ingestion+Lambdaプロセッサでのベクトル化とインデクシング 1万件の差分更新で表面化した問題 再設計：ベクトル化Lambdaを前段に出す 最終設計：S3+SQS+Lambdaで非同期にベクトル化とインデクシング ベクトル化Lambdaでの工夫 Bedrockのリージョン分散 1ファイル単位の処理量を制御する 出力形式と後処理 OpenSearch投入Lambdaでの工夫 external versionで古いデータで新しいデータを上書きすることを防止 処理完了後のファイル削除 Lambdaエラー時のファイル退避 非同期処理の完了待機 既存データに対する初回ベクトル化 結果 まとめ 背景 WEARでは、検索基盤として Amazon OpenSearch Service（以下、OpenSearch） を利用しています 1 。これまでフリーワード検索ではタグマッチングを主軸としていましたが、タグが付与されていない検索ワードに対する検索結果の質と量に課題がありました。 これを改善するため、ベクトル検索と全文検索を組み合わせたハイブリッド検索（WEARではあいまい検索と呼んでいるため、以下「あいまい検索」と表記）をリリースすることになりました 2 。 あいまい検索のためにベクトル検索を導入するには、検索対象の各documentに対して、タイトル・説明文・タグなどを連結したテキストをベクトル化したフィールドを持たせる必要があります。しかしながら、既存のインデクサーでこのフローを実現するのは難しく、インデクサー自体を刷新することになりました。本記事ではその刷新の過程と、設計時に行った工夫を紹介します。 既存のインデクサーと刷新の動機 WEARではOpenSearchへのインデクシングを Embulk を用いて行っていました。 embulk-input-bigquery と embulk-output-elasticsearch などを組み合わせ、 BigQuery からOpenSearchへデータを連携する構成です。Embulkのジョブは Digdag のworkflowで管理し、 Amazon EKS（以下、EKS） 上のJobとして実行していました 3 。インデクサーには差分更新と日次更新の2種類があり、それぞれ次の役割を持っていました。 差分更新：10分間隔で実行。直近で新規投稿・更新documentをインデクシングし、削除された投稿をindexから削除 日次更新：1日1回、新しいindexを作成して全件をインデクシングし、Blue/Greenでエイリアスを切り替える形で全件更新する。統計データなどの日次で更新すべき値はこのタイミングで反映 しかし、ベクトル検索の導入を検討するにあたり、この構成にはいくつかの課題がありました。 WEARで一番大きいコーディネートのindexは大量のdocumentを持っており、これらを毎日ベクトル化するのはコストと処理時間の両面で非現実的 BigQueryからOpenSearchへの連携中にベクトル化の処理を挟むのが困難 本対応の検討時点でEmbulkはすでにメンテナンスがされていない状態であり、長期的な保守性に不安 これらを踏まえ、ベクトル検索対応に必要な機能と、長期的な保守性の両方を満たす構成へとインデクサーを刷新する方針を決めました。 ベクトルデータの保持方法の検討 最初に取り組んだのが、ベクトルデータをどこに、どのタイミングで持たせるかという検討です。 既存の日次更新では新しいindexを毎日作成して全件インデクシングしていましたが、大量のデータを毎日全件ベクトル化するのは非現実的なため、ベクトルデータを別ストレージに保存しておく案を検討しました。しかし、ベクトル取得時のパフォーマンスやコスト面で見合わないと判断し、最終的には日次での全件更新そのものを廃止する方針を取りました。 毎日indexを全件更新するメリットの1つとして、indexの不整合が発生した場合に、日次での全件更新によって整合性を保つことができるという点がありました。これは例として、差分更新の失敗時のリトライで、古いデータで新しいデータが上書きされてしまうといった状況が挙げられます。全件更新を廃止するにあたり、この点をどう担保するのかが課題でしたが、後述する方法でdocumentのバージョニングを行うことで、不整合が発生しないようにしました。 新しい設計では、ベクトル化は差分更新のみで行い、日次更新では同じindexに対して日次で更新すべき値のみを上書きするように責務を分けました。これにより、ベクトル化を投稿の追加・更新時のみに限定でき、ベクトル化コストと処理時間の問題を回避できるようになりました。 インデクサーの構成方針 インデクサー刷新にあたって、ベクトル化を含む新しい構成として複数の選択肢を検討しましたが、OpenSearch Ingestionを軸とする構成を採用しました。判断のポイントは以下の通りです。 自前で運用する外部ツールは最小限にしたい（Embulkのように追加でメンテナンスが必要なツールを増やしたくない） データ抽出のSQLはバックエンドエンジニア、インデクサーのインフラ構築・運用はSREという責務分離を維持し、両者を疎結合にしたい AWS公式の Lambdaプロセッサでベクトル化するパターン を参考にすれば、ベクトル化部分をLambdaへ切り出して柔軟に構成できそう これらを総合的に考慮し、 Amazon S3（以下、S3） を起点とした AWS Lambda（以下、Lambda） の構成を方針として進めることになりました。 BigQuery → S3 のデータ連携 WEARでは Microsoft SQL Server からBigQueryへリアルタイム連携をしており、インデクサー側もBigQueryからデータを取得しています。前述の通り、インデクサーはS3を起点としてデータを処理する設計を取っているため、BigQueryから取得したデータをS3に連携する必要があります。BigQueryから直接S3へ出力する機能はないため、いったん Cloud Storage（以下、GCS） へ出力してからS3へ転送する形を取りました。 GCSへの出力にはBigQueryの EXPORT DATA 文 を利用しています。差分更新・日次更新いずれもJSON Lines形式でGCSへ出力するように記述しており、以下に差分更新を例にしたものを記載します。 EXPORT DATA OPTIONS( uri= ' gs://GCS_BUCKET/coordinates/diff/raw-data/YYYY/MM/DD/HH/mm/data_*.jsonl ' , format= ' JSON ' , overwrite= true ) AS -- 対象データを取得するクエリ ... uri にワイルドカード（ * ）を含めることで、BigQueryが出力サイズに応じて自動的に複数ファイルへ分割します。出力フォーマットは JSON を指定するとJSON Lines形式になります。 GCSからS3への転送方法は、差分更新と日次更新で異なるツールを使い分けています。 差分更新： rclone 日次更新： AWS DataSync（以下、DataSync） DataSyncは大量データの高速転送に適していますが、タスクの起動・実行に約5分かかります。差分更新は10分間隔で実行する上、ベクトル化のような時間のかかる処理も挟まるため、起動に時間のかかるDataSyncは許容できませんでした。差分更新ではデータ量がそこまで多くないこともあり、rcloneを採用しています。 日次更新の設計 日次更新では、もともとEmbulkで実装されていた全件更新を廃止し、差分更新と同じindexに対して統計データなどの日次で更新すべき値のみを上書きする方式に変更しました。日次更新の構成は以下の通りです。 日次更新はOpenSearch Ingestionを採用しており、S3に格納された全件データに対して S3 scan でOpenSearchへbulkでupsertしています。OpenSearch Ingestionのパイプライン定義の例は以下のとおりです。 version : 2 coordinates-daily-indexer : source : s3 : acknowledgments : true delete_s3_objects_on_read : true scan : buckets : - bucket : name : ${BUCKET_NAME} filter : include_prefix : [ "coordinates/daily/raw-data/" ] aws : region : ap-northeast-1 sts_role_arn : ${STS_ROLE_ARN} codec : ndjson : {} processor : - delete_entries : with_keys : - s3 sink : - opensearch : hosts : - https://${OPENSEARCH_HOST} aws : region : ap-northeast-1 sts_role_arn : ${STS_ROLE_ARN} index_type : custom index : coordinates document_id : ${/id} action : upsert max_retries : 10 bulk_size : 5 dlq : s3 : bucket : ${BUCKET_NAME} key_path_prefix : "dlq/coordinates/daily/" region : ap-northeast-1 sts_role_arn : ${STS_ROLE_ARN} source.s3.scan でS3バケット内の対象プレフィックスをスキャンします。 processor.delete_entries でS3イベントメタデータを落とし、 sink.opensearch でOpenSearchへbulk upsertしています。失敗したdocumentは dlq.s3 で指定したS3パスへ退避されます。 刷新前は4〜6時間ほど動き続けていた日次更新のジョブが、刷新後は1時間以内で完了するようになりました。これは更新フィールドを必要なものに絞れたことも要因の1つですが、OpenSearch Ingestionの処理が速いことも大きな要因です。 差分更新の設計 日次更新のような、すでにS3に格納されているデータをスキャンしてまとめて投入するユースケースでは、OpenSearch Ingestionは安定して動作することが分かっていました。ただし、S3 scanでS3データを処理するのはOpenSearch Ingestion起動のタイミングのみで、起動後にS3へ投入されたデータは処理されません。 日次更新の場合は実行のたびにOpenSearch Ingestionを起動し、完了したら終了させることで意図した動作が行えます。しかし、OpenSearch Ingestionの起動・終了にはそれぞれ5分ほどかかるため、10分ごとに実行される差分更新ではその実行時間は許容できません。 そのため差分更新では起動済みのOpenSearch Ingestionに Amazon Simple Queue Service（以下、SQS） 経由でデータを連携することにしました。これは、OpenSearch Ingestionのパイプラインを起動したまま、 SQS経由でリアルタイムに少量ずつデータを取り込む構成 です。 初期設計：OpenSearch Ingestion+Lambdaプロセッサでのベクトル化とインデクシング 差分更新の初期構成は以下の通りです。しかし、この構成ではいくつかの課題が発生し、最終的には断念しました。 OpenSearch Ingestionには Lambdaプロセッサ があり、パイプラインの途中でLambdaを呼び出して任意の処理を実行できます。これを使って Amazon Bedrock（以下、Bedrock） でのベクトル化を行う想定でした。 1万件の差分更新で表面化した問題 この構成で約1万件規模の差分更新を試したところ、いくつかの問題に直面しました。 OpenSearch IngestionのLambdaプロセッサは同期実行のみで、OpenSearch IngestionからLambdaへのread timeoutも10秒固定で調整できない 短時間に大量のベクトル化を行うとBedrockのリクエスト数クォータ超過で ThrottlingException が発生し、Lambda内でリトライしてもread timeoutする LambdaプロセッサがエラーになってもOpenSearch Ingestion側からリトライを設定する手段がない Lambdaプロセッサで処理が失敗したメッセージは OpenSearch IngestionのDead Letter Queue（以下、DLQ） には送られない。さらにS3の元ファイルも削除される仕様のため、失敗データが完全に消えてしまう。原因追跡やリカバリーができず、運用に耐えない パイプライン側からLambdaへの流量制限ができず、Bedrockを呼ぶLambdaへ過剰なリクエストが流れてしまう スロットリングを抑える目的でOCUと source.s3.sqs.maximum_messages を1にしても、Lambdaへの接続時にコネクションプール枯渇エラーが発生。S3に投入した1万件のうちOpenSearchに格納されたのは約6,900件で、残り3,100件ほどが毎回OpenSearch IngestionのDLQに溜まってしまう ここまでの検証から、OpenSearch IngestionのLambdaプロセッサは今回のユースケースには適さないと判断しました。 Lambdaプロセッサを諦めてベクトル化処理をOpenSearch Ingestionの外で行うことにします。後段のOpenSearch IngestionはS3に置かれたベクトル化済みデータを読んでSQS経由でリアルタイムに投入する構成とすれば、ここまでに挙げた問題は回避できそうだと考えました。 再設計：ベクトル化Lambdaを前段に出す OpenSearch IngestionからLambdaプロセッサを外し、ベクトル化を前段のLambdaに切り出した再設計を検証しました。しかし、この構成でも問題が発生しました。 ベクトル化部分をLambdaに切り出したことで、リトライ・流量制御をLambda側で完結できるようになり、初期設計で発生していた問題は解消されました。1万件規模の差分更新ではこの構成で安定して動作することを確認できました。 しかし、データ量を増やして検証を進めると、後段の SQS → OpenSearch Ingestion → OpenSearch の経路で別の問題が表面化しました。 3万件〜10万件規模になると、投入自体は完了しているにもかかわらず、OpenSearch IngestionがSQSメッセージを掴んだまま処理を継続し続け、可視性タイムアウトが切れて再処理が走る OpenSearch側でcircuit breakerが頻発し、 upsert 自体は成功しているにもかかわらずSQSメッセージが消費されず、同様に可視性タイムアウト切れで再処理が走る SQS → OpenSearch Ingestion の経路で流量制限ができないため、上記の挙動を緩和する手段がない 普段の差分更新で扱うデータ量が常に1万件以下に収まるなら見送りもできましたが、障害やメンテナンス明けに溜まったデータを一度に流すケースを考えると、大量データで挙動が崩れる構成は採用できません。 今回のユースケースだとOpenSearch Ingestionでは運用に耐えないと判断し、OpenSearch投入部分もLambdaに置き換える方針としました。 最終設計：S3+SQS+Lambdaで非同期にベクトル化とインデクシング 再設計で残っていたOpenSearch Ingestion部分も自作のLambdaに置き換え、S3とSQSを挟んで非同期に処理を進める構成にしました。最終的に、この構成で安定して処理できるようになりました。 ベクトル化を担うLambdaと、OpenSearchへ投入するLambdaを分け、それぞれS3のオブジェクト作成イベントをSQS経由で受け取って動作させます。この構成にすることで、次のような恩恵が得られました。 ベクトル化でBedrockのクォータを超過した場合、SQSが自動的にリトライする Lambdaの同時実行数を設定することで、BedrockとOpenSearchへの流量を独立に制御できる ベクトル化処理が重くなっても、後段のOpenSearch投入には影響が及ばない この構成に切り替えた後、10万件規模の差分更新でも想定時間内に推論からupsertまで完了することを確認でき、再設計で発生していたSQS再処理やcircuit breakerの問題も解消されました。実運用の定常時は1万件以下、最大でも数万件規模ですが、障害やメンテナンス明けに大量データを一度に流すケースでも問題なく捌けるようになっています。 ベクトル化Lambdaでの工夫 ベクトル化LambdaはS3に格納されたファイルをSQS経由で受け取り、1ファイルずつ処理します。次のような点を工夫しました。 Bedrockのリージョン分散 ベクトル化に利用している amazon.titan-embed-text-v2:0 モデルは、リージョンごとに1分あたり6,000回のクォータがあり、これは調整不可でした 4 。WEARでは OpenSearchのコネクタ 経由でも検索時の検索ワードのベクトル化を行っており、こちらは ap-northeast-1 リージョンを利用しています。インデクサーのベクトル化が ap-northeast-1 のクォータを使い切ってしまうと、検索時のベクトル化に影響が出てしまいます。 そのため、ベクトル化Lambdaでは検索とは別のリージョンを使うようにしました。具体的には us-east-1 と us-west-2 の2つを使い、リクエストごとに順序をシャッフルして順に試行する形にしています。これにより、検索側のクォータに影響を与えず、かつLambda側で利用できるクォータも実質的に2倍に増やせています。 クォータ超過（ ThrottlingException ）の場合は別リージョンへフォールバックし、すべて埋まっていたら指数バックオフでリトライするようにしました。 1ファイル単位の処理量を制御する ベクトル化Lambdaは1ファイル1実行で動作するため、1ファイル内のデータ量が多すぎるとBedrock呼び出しに時間がかかり、Lambdaのタイムアウトに引っかかる問題が発生しました。Lambdaのタイムアウトはリトライまでの時間とBedrockクォータエラー時の挙動を考慮して60秒に設定しています。 最初はBigQueryのEXPORT DATA時点でファイルサイズを制限する方法を検討しました。公式ドキュメントの エクスポートファイルのサイズを制限する に従い、対象データをパーティション分割した上でEXPORT DATAをループ処理する方式です。しかし、この方式はBigQuery側の並列ワーカーで一括出力する場合と比べて出力時間が大幅に伸びてしまい、差分更新の実行間隔である10分には到底収まらないため断念しました。 最終的には、GCSからS3へ転送する段階で1ファイルあたり500KBを上限として分割し、1ファイル内のデータ量を一定以下に抑えました。シェルスクリプトとして以下のような処理を組み込んでいます。 # GCS から S3 へのストリーミング転送例 # 第1引数: SRC（例: gcs:bucket/prefix/）, 第2引数: DST（例: s3:bucket/prefix/） SRC = " $1 " ; DST = " $2 " CHUNK_BYTES = " 500K " JOBS = 4 # 1) しきい値以下のファイルはそのまま move rclone move " $SRC " " $DST " \ --max-size " $CHUNK_BYTES " --min-size 1B --include " *.jsonl " # 2) しきい値を超えるファイルは rclone cat → split で分割しつつ、 # 分割ファイルを rclone rcat でそのまま S3 へ PUT（中間ファイルを作らない） rclone lsf -R --files-only --min-size " $CHUNK_BYTES " " $SRC " \ | tr ' \n ' ' \0 ' \ | xargs -0 -P " $JOBS " -I{} bash -euo pipefail -c ' REL="$1" STEM="${REL##*/}"; STEM="${STEM%.*}" rclone cat "${SRC}${REL}" \ | split -C "${CHUNK_BYTES}" - chunk- \ --numeric-suffixes=1 --suffix-length=5 \ --additional-suffix=.jsonl \ --filter "rclone rcat \"${DST}${STEM}-\$(basename \$FILE)\"" rclone deletefile "${SRC}${REL}" ' bash " {} " split の --filter オプションを使うことで分割ファイルをディスクに書き出さず、 rclone rcat で直接S3へPUTできます。これによりPodの一時ストレージを使い切らずに済み、 xargs -P によって並列実行することで転送時間も抑えています。 ファイル分割により、1ファイルあたりの処理時間がLambdaのタイムアウト以内に収まり、ベクトル化Lambdaが安定して動作するようになりました。 出力形式と後処理 ベクトル化Lambdaは、JSONから対象テキストを取り出してベクトル化し、元のフィールドの代わりにベクトル用フィールドを追加した上で、JSON Lines形式でS3の別パスに出力します。OpenSearch投入用Lambda側のSQSがそのパスのS3作成イベントを受け取るようにし、後続の処理を行います。出力完了後は元ファイルを削除します。 OpenSearch投入Lambdaでの工夫 OpenSearch投入Lambdaは、ベクトル化済みデータが格納されたS3パスをSQS経由で受け取り、適切なバッチサイズにまとめてOpenSearchへ投入します。 external versionで古いデータで新しいデータを上書きすることを防止 差分更新の全体はDigdag workflowで管理していますが、WEAR全体のメンテナンスやインデクサーのエラーなどでジョブを再実行する可能性があります。その際、後の時間帯のデータを取得したジョブの後に、前の時間帯のデータを取得したジョブを実行する可能性があり、古いデータで新しいデータを上書きしてしまう懸念がありました。 これを防ぐため、OpenSearchの Bulk API で version / version_type を渡し、external versionによって更新可否を判定できるようにしています。データ取得範囲の終了時刻をunixtimeに変換した値をversionとして持たせ、現在のversion以上の値の場合のみ更新が成立するようにしました。これにより、ジョブの実行順序が前後しても古いデータで上書きされることはなくなります。 unixtimeはOpenSearch投入Lambdaに渡されるS3キーのパスから算出しています。ベクトル化Lambdaが出力するS3キーはデータ取得範囲の終了時間を含んだ階層構造であり、それをパースしてversionとしています。 from opensearchpy import helpers # OpenSearchへのbulk投入例（簡略化） # 各アクションのメタデータに version / version_type を含めることで、 # external_versionより小さいversionの更新を拒否させる # 更新処理にバグ等があった場合に再度同じ時間帯のデータで更新できるように、 # version_typeはexternal_gte（現在のversion以上のときのみ更新）を指定している actions = [ { "_op_type" : "index" , "_index" : index_name, "_id" : record[ "id" ], "_source" : record, "_version" : external_version, "_version_type" : "external_gte" , } for record in records ] helpers.bulk(client, actions) なお、external versionはupsertに対応していないため 5 、差分更新ではdocument全体をreplaceする形で更新しています。 日次更新のOpenSearch Ingestionではexternal versionを利用できませんが、upsertすることでversionが+1される形で更新されます。 差分更新と日次更新でデータ投入のタイミングが前後する可能性はありますが、日次で更新する値は仮に古い値で更新されても致命的な問題にはならない内容のため許容しています。 処理完了後のファイル削除 OpenSearchへの投入が完了したファイルはS3から削除します。これは後述する非同期処理の完了待機の仕組みのためでもあります。 Lambdaエラー時のファイル退避 差分更新のLambdaでエラーが発生し、SQSの最大リトライ回数を超えた場合は error/ パスへファイルを移動させています。これにより後述する完了待機が止まらないようにしつつ、後でエラーになったファイルを追跡できるようにしています。 ただし、Lambdaがタイムアウトで終了した場合は error/ パスへの移動前に終了してしまいます。その結果、SQSのDLQにメッセージが残ったまま、元ファイルがS3に残ったままとなり、Digdag workflowの完了待機が永遠に終わらなくなります。 これを補完するため、定期的にSQSのDLQを監視し、 error/ パスへ移動できていないファイルを検知して移動する補助Lambdaを別途用意しています。 非同期処理の完了待機 インデクサーには非同期処理が含まれており、Digdag workflow側からは処理がいつ終わったかを直接知る術がありません。そこで、Digdag workflow側では処理対象のS3パスを監視し、ファイルが残っていない状態になったら処理完了とみなす方式を取りました。完了待機を入れる理由は次の通りです。 Bedrockのクォータで詰まっている状況で次のジョブが走ると、SQSにファイルがどんどん溜まってしまう すべての投入が完了したタイミングでindex refreshを呼び、検索に反映させる必要がある refresh頻度が高いと日次更新時のCPU使用率に影響するため、自動refreshは無効化している aws s3 ls コマンドでS3の対象パスをリストし、ファイルが存在しなければ処理完了とみなすロジックをシェルスクリプトで実装しています。 既存データに対する初回ベクトル化 新しいインデクサーをリリースする前に、既存の大量のdocumentにベクトルデータを付与する必要がありました。差分更新と日次更新のジョブを稼働させ始めるだけでは、その時点で過去に投稿された大量のdocumentにはベクトルデータが付かないままです。 そこで、初回ベクトル化用に AWS Step Functions を作成し、次の手順で全データのベクトル化を実施しました。 Step Functionsで全データのバッチベクトル化を実行 OpenSearch Ingestionで全データをOpenSearchへ投入 差分更新・日次更新のジョブを稼働開始 1〜3の処理中に取りこぼした時間帯のデータを、差分更新のジョブで埋める external versionの仕組みがあるため、4の時点で過去の時間帯のデータを後から流しても、すでに最新のデータが入っているdocumentが古いデータで上書きされてしまうことはありません。差分更新の設計時点で順序を気にしなくて良いようにしたことが、初回ベクトル化のフローでも生きました。 結果 Embulkを用いた構成から、OpenSearch Ingestionと自作Lambdaを組み合わせた構成へとインデクサーを刷新したことで、以下のような効果が得られました。 日次更新のジョブが4〜6時間から1時間以内に短縮された 10万件規模の大量データを流すケースでも、想定時間内に推論からupsertまで完了できるようになった Embulkが メンテナンスモード になるタイミングと重なり、結果的にリプレイスも同時に行えた まとめ 本記事では、WEARのあいまい検索リリースに伴って検索インデクシングシステムを刷新した話を紹介しました。差分更新ではOpenSearch IngestionのLambdaプロセッサで要件を満たせなかったため、S3とSQSを挟んで自作Lambdaで非同期に処理を進める構成に切り替えました。日次更新ではOpenSearch IngestionのS3スキャンを採用し、毎回起動・終了させるという形で、ユースケースに応じて使い分けました。 ベクトル検索を支えるバッチ処理を設計する上では、以下のような工夫が有効でした。 ベクトル化用のBedrockをアプリケーションとは別リージョンに逃がし、クォータを分離する ファイル単位の処理量をあらかじめ転送段階で制御し、Lambdaの処理時間を短縮させる external versionでジョブ実行順序の前後に対する耐性を持たせる ベクトル化と投入をLambdaとして分割し、流量制御や障害の影響範囲を限定する OpenSearch Ingestionの採用やベクトル検索のためのインデクサーの設計を検討している方の参考になれば幸いです。 ZOZOでは、一緒にサービスを作り上げてくれる方を募集中です。ご興味のある方は、以下のリンクからぜひご応募ください。 corp.zozo.com 元々はElasticsearchを使用していましたが、OpenSearchへ移行しました。詳細は WEARの検索基盤をElasticsearch 7.10.2からOpenSearch 2.19.0へ無停止で移行する ── ダブルライトとカナリアリリースによる段階的アプローチ をご参照ください。 ↩ あいまい検索のリリースに関する詳細は別記事で紹介予定です。 ↩ 実際はフォークしてカスタマイズしたプラグインを使用しています。 ↩ クォータの値は本記事の執筆時点のものです。最新の値は Amazon Bedrock サービスクォータ一覧 をご参照ください。 ↩ Bulk - OpenSearch Documentation 。 version_type=external 系を指定した場合、documentの作成または完全置換のみが対象になります。 ↩

はじめに こんにちは、WEAR開発部バックエンドブロックの小山です。普段は弊社サービスである WEAR のバックエンド開発を担当しています。 WEARではハイブリッド検索などの新たな検索体験の実現を目指しています。その実現に必要な ハイブリッド検索 はOpenSearch 2.11で導入された機能です。Elasticsearch 7.10.2では利用できないため、Amazon OpenSearch Service上のエンジンをOpenSearch 2.11.0以上へ移行する必要がありました。今回はOpenSearch 2系の最新バージョンだった2.19.0を採用しました。本記事では、この移行にあたり対応したSearchkickの導入、ダブルライト戦略によるインデクシング移行、カナリアリリースによる段階的トラフィック切り替えについてご紹介します。 目次 はじめに 目次 抱えていた課題 Elasticsearch 7.10.2の限界 既存のアーキテクチャ 課題を解決したアプローチ 1. Searchkickとopensearch-rubyへの移行 elasticsearch-modelからSearchkickへ elasticsearchからopensearch-rubyへ 既存Searchableとの並存 2. インデクシングのダブルライト戦略 embulk-outputの変更 RakeタスクとDigdagワークフローの追加 3. クエリ種別ごとの動作確認 確認の目的と方針 確認対象の抽出方法 確認したクエリ種別 確認方法 4. 負荷試験 試験条件 試験結果 5. カナリアリリースによる段階的トラフィック移行 リリーススケジュール 各段階での確認項目 確認結果 効果と得られた知見 移行後のアーキテクチャ Searchkickとopensearch-rubyへの移行による保守性向上 並行稼働時のインデクサー移行方法 カナリアリリースの有効性 おわりに 抱えていた課題 Elasticsearch 7.10.2の限界 WEARではコーディネートや動画、メイクの投稿検索にAmazon OpenSearch Service上でElasticsearch 7.10.2を利用していました。しかし、以下の課題がありました。 新機能の利用不可：WEARではハイブリッド検索などの新たな検索体験を計画していたが、Elasticsearch 7.10.2はハイブリッド検索に対応しておらず、実現できない状態 サポートの先行き不透明：Elasticsearch 7.10.2は、Amazon OpenSearch Serviceで提供される最終のオープンソースElasticsearchバージョン。今後の新機能追加やセキュリティパッチの提供が見込めない状態。Elasticsearch 7.1〜7.8の標準サポートは2025年11月に終了しており、7.10.2も同様のサポート終了が予想される状態。AWS側でもOpenSearchエンジンへの移行を推奨 ライブラリのメンテナンス性： elasticsearch gem 7.14.0以降ではAmazon OpenSearch Service上のElasticsearchへ接続不可。gemのバージョンを7.13.3に固定せざるを得ず、アップデートができない状態 既存のアーキテクチャ WEARの検索基盤は、以下のシステム構成で運用していました。 検索機能： elasticsearch-model gemを利用し、検索メソッドを提供。内部では elasticsearch gemが提供する Elasticsearch::Client を通じてOpenSearch Serviceと通信 マッピング定義： elasticsearch-model gemを利用し、モデルにマッピング定義を記述 インデックス操作： elasticsearch gemを利用し、Rakeタスクによるインデックス作成、エイリアス切り替え、旧インデックス削除、ドキュメント削除 インデクシング：トラフィックを考慮し、レコード更新ごとではなくDigdagワークフローと Embulk による定時バッチ（日次の洗い替えと差分更新）でインデクシング 課題を解決したアプローチ 今回の移行では、既存ドメインのインプレースアップグレードではなく、OpenSearch 2.19.0の新規ドメインを作成し、エンドポイントを段階的に切り替える方法を採用しました。その理由は以下の通りです。 インプレースアップグレードでは、Elasticsearch 7.10.2からOpenSearch 2.19.0へ直接移行できず、 OpenSearch 1.xを経由する必要がある elasticsearch-model / elasticsearch から searchkick / opensearch-ruby へのgem移行が必要であり、アプリケーションコードに破壊的変更が生じる 検索基盤は影響範囲が大きいため、カナリアリリースで段階的にリリースしたい これらを踏まえ、Elasticsearchをダウンタイムなく移行させるために以下のアプローチで段階的に進めました。 Searchkickとopensearch-rubyへの移行 インデクシングのダブルライト戦略 クエリ種別ごとの動作確認 負荷試験 カナリアリリースによる段階的トラフィック移行 1. Searchkickとopensearch-rubyへの移行 移行前後のgemの対応関係は以下の通りです。 責務 Elasticsearch利用時 OpenSearch移行後 検索機能 elasticsearch-model （内部で elasticsearch を利用） searchkick （内部で opensearch-ruby を利用） マッピング定義 elasticsearch-model searchkick インデックス操作 elasticsearch 直接利用 opensearch-ruby 直接利用 elasticsearch-modelからSearchkickへ 検索機能とマッピング定義については、既存の elasticsearch-model の代わりに、 searchkick に移行しました。Searchkickを選定した理由は以下の通りです。 OpenSearchを公式にサポートしている リポジトリが継続的にメンテナンスされている nested型への対応など、 elasticsearch-model との互換性がある reindex時のアトミックなエイリアス切り替えが組み込まれているほか、ハイブリッド検索やセマンティック検索にも対応しており、高度な機能を備えている elasticsearchからopensearch-rubyへ インデックス操作のRakeタスクでは、 elasticsearch を使用していました。OpenSearch移行に伴い、これを opensearch-ruby に置き換えました。 - require 'elasticsearch' - client = Elasticsearch::Client.new(client_options) + require 'opensearch-ruby' + client = OpenSearch::Client.new(client_options) client.indices.update_aliases(...) client.indices.delete(...) opensearch-ruby は elasticsearch とAPIの互換性が高いため、クライアントの初期化部分とエラークラスの変更で、既存のインデックス操作ロジックをそのまま利用できました。 唯一の例外がインデックス作成タスクで、ここではSearchkick経由でマッピング定義を取得して作成しています。 task :create_index , [ :index_name ] => :environment do |_, args| index_class = index_class_name(args[ :index_name ]).singularize.capitalize.constantize index = Searchkick :: Index .new(args[ :index_name ]) model_config = index_class.search_index.index_options # Searchkickからマッピング取得 index.create(model_config) # Searchkick経由で作成 end このように、マッピング定義はSearchkickに一元化しつつ、その他のインデックス操作は opensearch-ruby を直接使用する構成としました。 既存Searchableとの並存 WEARでは、モデルごとに *Searchable というconcernを定義し、 elasticsearch-model を利用した検索用のデータ定義とマッピング定義を集約していました。 移行期間中は、Elasticsearchを利用するサーバーとOpenSearchを利用するサーバーを並行稼働させる必要がありました。そこで、モデルごとに *OpensearchSearchable concernを新設し、既存の *Searchable と並存させる構成をとりました。 既存の *Searchable はElasticsearch用のconcernです。 # 既存: Elasticsearch用 module Searchable extend ActiveSupport :: Concern # elasticsearch-model を利用したデータ定義とマッピング定義 end 新設した *OpensearchSearchable はOpenSearch用のconcernです。 # 新規: OpenSearch用 module OpensearchSearchable extend ActiveSupport :: Concern included do searchkick index_name : Rails .configuration.x.application[ :opensearch ][ :index_name ], settings : Rails .configuration.x.application[ :opensearch ][ :settings ], callbacks : false , merge_mappings : true , mappings : search_mappings def search_data # searchkick を利用したデータ定義 end end module ClassMethods def search_mappings # searchkick を利用したマッピング定義 end end end merge_mappings: true を指定することで、独自に定義したマッピングをSearchkickの自動生成マッピングにマージしています。 callbacks: false を指定することで、Searchkickの自動インデクシングを無効化し、既存のEmbulkによるインデクシングとの競合を防いでいます。 2. インデクシングのダブルライト戦略 移行期間中、ElasticsearchとOpenSearchの両方にデータを投入するダブルライトを実施しました。WEARのインデクシングは日次バッチによる洗い替え方式のため、ダブルライトを開始した時点で既存データも含めてOpenSearchに自動で同期されます。そのため、既存データの移行作業を別途行う必要はありませんでした。 embulk-outputの変更 前述の通り、既存の構成ではEmbulkを介して、BigQueryからデータを取得してElasticsearchにインデクシングしていました。インデクシング時のBigQueryのクエリコストが高額なため、OpenSearchにもインデクシングを行う際に単純にジョブを複製してしまうと、費用が2重に掛かってしまうという課題がありました。 そこで、embulk-outputの出力先をElasticsearchとOpenSearchの両方に向けることで、SQLの実行は一度だけで双方にデータを転送できるようにしました。 移行前はElasticsearchのみに出力していました。 # Elasticsearchへのインデクシング時 out : type : elasticsearch mode : insert nodes : - { host : {{ elasticsearch_host }} , port : {{ elasticsearch_port }}} index : {{ elasticsearch_index }} { % Elasticsearchの設定値 % } ダブルライト時は type: multi を使い、ElasticsearchとOpenSearchの両方に出力しました。 # ElasticsearchとOpenSearchにダブルライトするインデクシング時 out : type : multi outputs : - type : elasticsearch mode : insert nodes : - { host : {{ elasticsearch_host }} , port : {{ elasticsearch_port }}} index : {{ elasticsearch_index }} { % Elasticsearchの設定値 % } - type : elasticsearch mode : insert nodes : - { host : {{ opensearch_host }} , port : {{ opensearch_port }}} index : {{ opensearch_index }} { % OpenSearchの設定値 % } ダブルライトのために embulk-output-multi を新たに導入し、複数出力先への分岐を実現しました。OpenSearch側の出力も type: elasticsearch を指定しています。 embulk-output-elasticsearch はOpenSearchとのAPI互換性により、そのままOpenSearchへの出力にも利用できました。 RakeタスクとDigdagワークフローの追加 OpenSearch向けのインデックス操作のRakeタスクとDigdagワークフローを作成し、OpenSearchに対しても実行できるようにしました。 # 既存のElasticsearchのインデックス作成 +create_index_elasticsearch: sh>: ... rails "elasticsearch:create_index[${index_name}]" # 追加したOpenSearchのインデックス作成 +create_index_opensearch: sh>: ... rails "opensearch:create_index[${index_name}]" 3. クエリ種別ごとの動作確認 OpenSearch移行後にすべてのクエリ種別が正常に動作するかをQA環境で確認しました。 確認の目的と方針 Elasticsearchに送信されるクエリの種別ごとに、OpenSearch上でも同等の結果が返ることを確認しました。クエリ種別が重複するエンドポイントは確認対象外とし、効率的に網羅性を担保しました。 確認対象の抽出方法 確認対象の抽出は以下の手順で行いました。 対象エンドポイントの洗い出し：リポジトリ内でElasticsearchのQueryクラスを呼び出している箇所をリストアップ WEAR Webの対象画面の特定：Webマスタ仕様書から対象エンドポイントが使用されている画面を確認 クエリの特定：APIのリクエストパラメーターから生成されるOpenSearchのクエリJSONを特定し、使用されているクエリ種別を分類 確認したクエリ種別 以下のクエリ種別を対象に、WEAR iOS・Android・Webの各プラットフォームで動作確認を実施しました。 分類 クエリ種別 検索クエリ term 、 terms 、 range 、 nested 、 bool （ filter / must_not / must / should ）、 function_score 、 exists ソート sort ページング from 、 size グループ化 collapse 複合検索 msearch 確認方法 WEAR iOS・Android・Webの各プラットフォームで、以下の方法で確認しました。また、対応するRSpecテストを実行し、OpenSearchに対するクエリが正常に動作することはCI上で確認しています。 WEAR iOS・Android：QA環境のAPIに対してcurlコマンドでリクエストを送信し、レスポンスを確認。 WEAR Web：ブラウザ上で対象画面を操作し、APIレスポンスと画面表示を目視確認。 すべてのクエリ種別で正常な動作を確認し、負荷試験に進みました。 4. 負荷試験 本番リリース前に、OpenSearchクラスターがElasticsearch利用時と同等のリクエスト量を処理できるかを確認するため、QA環境で負荷試験を実施しました。 試験条件 QA環境のOpenSearchクラスターを本番環境のElasticsearchと同等のスペックに設定 検索エンドポイントのRedisキャッシュを無効化し、OpenSearchへの直接的な負荷を計測 k6を用いて、各検索エンドポイントに対して本番のピーク帯のMAX rps相当のリクエストを6時間継続 試験結果 レイテンシ ：Datadog APMで各検索エンドポイントのp99レイテンシを直近1か月の平均と比較した結果、OpenSearchがボトルネックとなるレイテンシ劣化は観測されなかった エラー ：Datadog APMで各検索エンドポイントを確認した結果、OpenSearch起因のエラーは発生しなかった クラスターメトリクス ：本番のピーク帯MAX値相当のリクエストを6時間継続した。CPUUtilizationはリクエスト量に対して許容範囲内、JVMMemoryPressureは本番環境と同程度であり、各種メトリクスに大きな影響はなかった この結果をもとに、カナリアリリースによる段階的な本番投入を判断しました。 5. カナリアリリースによる段階的トラフィック移行 本番リリースでは、カナリアリリースによって段階的にトラフィックを移行しました。 リリーススケジュール 日時 内容 2025/9/30 13:00 canary podの作成、APIの正常確認、1％リリース 2025/9/30 17:00 10％リリース 2025/10/1 14:00 50％リリース 2025/10/2 13:30 100％リリース 2025/10/2〜10/6 正常性の継続監視 各段階での確認項目 各段階で以下の項目を確認し、問題がなければ次の段階に進みました。 OpenSearchのレイテンシ比較とエラー確認：Datadog APMでOpenSearchとElasticsearchのレイテンシを比較し、劣化がないことを確認。OpenSearchのエラーがないことを確認。 各検索エンドポイントのレイテンシ比較とエラー確認：Datadog APMで各検索エンドポイントのレイテンシを比較し、劣化がないことを確認。OpenSearch起因のエラーがないことを確認。 クラスターメトリクス：SearchLatency、IndexingLatency、CPUUtilization、JVMMemoryPressureを監視し、劣化がないことを確認。 インデックスの整合性：ElasticsearchとOpenSearchのドキュメント件数に差異がないことを確認。 確認結果 OpenSearchでレイテンシが低い傾向を確認した（平均・最小・最大いずれもOpenSearchの方が高速） OpenSearch起因のエラーが発生しなかった OpenSearchでJVMMemoryPressureがやや高い傾向にあったが、MAXでも60％未満であり問題なかった CPUUtilizationはOpenSearchの方が低い傾向だった 100％リリース後の監視でも劣化が見られず、移行完了を判断した 効果と得られた知見 移行後のアーキテクチャ 移行後の検索基盤は、以下のシステム構成になりました。 検索機能： searchkick gemを利用し、検索メソッドを提供。内部では opensearch-ruby gemが提供する OpenSearch::Client を通じてOpenSearch Serviceと通信 マッピング定義： searchkick gemを利用し、モデルにマッピング定義を記述 インデックス操作： opensearch-ruby gemを利用し、Rakeタスクによるインデックス作成、エイリアス切り替え、旧インデックス削除、ドキュメント削除 インデクシング：既存のDigdagワークフローと Embulk による定時バッチ（日次の洗い替えと差分更新）でのインデクシングを継続 Searchkickとopensearch-rubyへの移行による保守性向上 elasticsearch-model から searchkick 、 elasticsearch から opensearch-ruby に移行し、以下の効果と知見がありました。 OpenSearchの将来的なバージョンアップへの追随が容易になった reindex処理のアトミックなエイリアス切り替えが組み込みで利用可能になった ハイブリッド検索の機能が利用可能になった opensearch-ruby はAPI互換性が高く、Rakeタスクの移行コストが低かった 並行稼働時のインデクサー移行方法 ダブルライト戦略により、以下のメリットがありました。 ElasticsearchとOpenSearchを並行稼働させることで、いつでも切り戻し可能な状態を維持 Embulkを利用した既存のインデクシングパイプラインを最小限の変更で拡張 移行時のクエリコスト増大を防止 Digdagワークフロー層での制御により、アプリケーションコードへの影響を最小化 カナリアリリースの有効性 段階的なトラフィック移行により、以下の知見が得られました。 1％リリースと10％リリースで、JVMMemoryPressureの変動が大きく見られた。これは、リリース後の低トラフィック時にキャッシュヒット率が低いことに起因する可能性が高く、50％リリース以降は安定した。 検索基盤のような影響範囲の大きいミドルウェアの移行にはカナリアリリースが有効であることを実感した。 おわりに 本記事ではWEARにおけるElasticsearch 7.10.2からOpenSearch 2.19.0への移行プロセスを紹介しました。同様の移行を検討している方の参考になれば幸いです。 ZOZOでは、一緒にサービスを作り上げてくれる方を募集中です。ご興味のある方は、以下のリンクからぜひご応募ください。 corp.zozo.com

はじめに こんにちは、商品基盤部の杉浦、小原、寺嶋です。普段はZOZOTOWNのお気に入り基盤・商品レビュー基盤といった商品サブドメインを担当しています。 私たちのチームでは運用コスト削減を目的として、お気に入りデータベースをオンプレミスのSQL ServerからAWS Aurora MySQLへの移行に取り組んでいます。お気に入りデータは数十億レコードに及び、移行中もデータが増え続けるためデータの静止点が作れないという課題がありました。本記事では、この大規模データ移行における初期移行の取り組みと、Embulkを用いた差分同期について紹介します。 なお、新規データの書き込みを担保するダブルライト戦略については 前回の記事 で紹介しています。あわせてご覧ください。 目次 はじめに 目次 お気に入りリプレイスの概要 技術スタックの老朽化 オンプレミスSQL Serverの運用限界 背景・課題 初期移行 制約と課題 検証と最適化 本番移行の結果 得られた学び Embulkによる差分同期 ジョブ設計 ソースDBへの負荷制御 データ整合性の担保 設定管理とチューニング まとめ お気に入りリプレイスの概要 ZOZOTOWNのお気に入り機能は、会員が興味のある商品・ブランド・ショップを登録し、お気に入り一覧から確認できる機能です。まず、ユーザー種別として 会員 と ゲスト会員 の2種類が存在し、それぞれ独立したテーブルで管理されています。お気に入り登録の対象も 商品・ブランド・ショップ の3種類があり、ユーザー種別との掛け合わせにより、合計6パターンのテーブルが移行対象となります。さらに、 過去に削除されたお気に入りの履歴（アーカイブデータ） も保持されており、これらを含めると移行対象のテーブルは多岐に及びます。テーブルによってレコード数は数千万レコードから数十億レコードまで幅があり、合計すると数十億レコード規模のデータ移行となりました。 この構成は長年にわたりZOZOTOWNを支えてきましたが、以下のような課題を抱えていました。 技術スタックの老朽化 ZOZOTOWNは2004年の開始当初からClassic ASP（VBScript）とSQL Serverのストアドプロシージャでビジネスロジックを実装してきました。しかし、VBScriptは開発元のMicrosoftも積極的に開発しておらず、クラウドベンダーのSDKが提供されていないなど技術的な制約が大きくなっていました。こうした背景からZOZOTOWN全体で リプレイスプロジェクト が進められており、お気に入り機能もその一環としてマイクロサービスへの刷新に取り組んでいます。 オンプレミスSQL Serverの運用限界 ZOZOTOWNは運営開始から10年以上にわたりオンプレミス環境でシステムを拡大してきましたが、スケーラビリティや保守コストの面で課題を抱えていました。2017年より ストラングラーフィグパターンによる段階的なマイクロサービス移行 が進められています。お気に入り機能のデータベースもその一環として、オンプレミスのSQL ServerからAWS上のAurora MySQLへの移行が必要でした。しかし、以下の制約がありました。 Read/Writeが常時発生しており、 システム停止を伴う移行は不可能 書き込んでから読み取れるまでの許容タイムラグが短く、 レプリケーション方式では要件を満たせない オンプレミスDBへの設定変更が必要なマネージドサービス（AWS DMS等）は、 他機能への影響を考慮し使用を見送り お気に入りデータが膨大なため、 インデックス設定などのチューニングにも数時間を要する状態 これらの課題を踏まえ、移行方式を設計し技術検証しました。移行戦略の全体像は以下の3フェーズで構成されています。 フェーズ1 : SQL Server単体での運用（移行前） フェーズ2 : SQL ServerとAurora MySQLのデュアル運用（移行期間） フェーズ3 : Aurora MySQL単体での運用（移行完了） フェーズ2におけるダブルライトの仕組みやフェーズ切り替えの実装については 前回の記事 で紹介しています。本記事ではこのフェーズ2にフォーカスします。 背景・課題 初期移行 初期移行は、ソースDB（オンプレミスSQL Server）からターゲットDB（Aurora MySQL）へのデータ一括移行です。全体の流れは以下の通りです。 抽出 : SQL Serverから bcp でCSV出力 転送 : CSVファイルをS3へアップロード ロード : LOAD DATA FROM S3 でAurora MySQLへインポート インデックス構築 : ALTER TABLE でインデックスを追加 制約と課題 今回の初期移行には、以下の制約がありました。 ソースDB（本番稼働中） : 影響を最小限に抑える必要がある ターゲットDB（サービスイン前） : 大胆な最適化が可能 この非対称な条件から、「 抽出は慎重に、インポートは大胆に 」という方針を採用しました。抽出には bcp （Bulk Copy Program）を採用しました。 bcp はSQL Server標準のバルクエクスポートツールであり、SELECT文による抽出と比較して以下の利点があります。 高スループット : 200,000〜500,000行/秒の安定した出力性能 シンプルな運用 : 追加のミドルウェアやライセンスが不要 転送ではS3を中継することで、ロード失敗時に再抽出せず再実行できる設計としています。 一方、事前試算では最大規模テーブルのインポートに 数日〜1週間 を要することが判明しました。ロード時間が長期化すると、以下のリスクが高まります。 接続切断・タイムアウト : 数日に及ぶ処理は中断リスクが高い 障害時の復旧困難 : 失敗時のデバッグと再実行に多大な時間を要する 移行スケジュールへの影響 : ダブルライト期間が長期化し、運用負荷が増大する ロールバック困難 : 問題発覚時に手戻りできる時間的余裕がなくなる これらのリスクを軽減するため、インポート処理の最適化が必須でした。 検証と最適化 本番移行に先立ち、約6,000万レコードを持つテーブルを用いて3つの観点で検証しました。 1. 並列化の効果 LOAD DATA FROM S3 MANIFEST でマニフェスト分割による並列実行を検証しました。CSVファイルを4分割・8分割・16分割と変化させましたが、スループットは 約51,000〜53,000行/秒で横ばい でした。 今回のAurora構成はProvisioned（単一ライターノード）であり、並列ロードを実行してもCPUおよびストレージI/O帯域がボトルネックとなります。Aurora Serverless v2のような動的スケーリング構成であれば結果が異なる可能性もありますが、今回の構成では並列化による改善は限定的でした。 2. インデックス戦略 方式 内容 処理効率 パターンA インデックスなしでLOAD → 後からALTERで追加 約61,000〜68,000行/秒 パターンB インデックスありでLOAD 約39,000〜42,000行/秒 パターンAが 最大59％高速 でした。行挿入ごとのインデックス更新はランダムI/Oを発生させますが、一括構築ならソート後、シーケンシャルに処理できます。ターゲットDBは未稼働のため、この最適化を採用しました。 3. インスタンスサイズ インスタンスタイプ別のスループットを比較しました。料金は Amazon Aurora の料金 を参照しています。 インスタンス インポート効率 ALTER効率 オンデマンド時間単価 r6i.2xlarge 約125,500行/秒 約120,300行/秒 約$0.63/時 r6i.16xlarge 約162,200行/秒 約162,800行/秒 約$5.00/時 r6i.16xlargeはr6i.2xlargeと比較して約30％のスループット向上が見られた一方、コストは約8倍です。このスループット差がテーブル規模によって処理時間に与える影響は以下の通りです。 大規模テーブル（数十億レコード） : 2〜3時間の短縮 → リスク低減に寄与 小規模テーブル（数千万レコード） : 数分の短縮 → コスト対効果が低い この結果から、大規模テーブルはr6i.16xlargeで時間短縮とリスク低減を図り、中小規模テーブルはr6i.2xlargeでコスト効率を最大化する ハイブリッド戦略 を採用しました。 本番移行の結果 検証結果をもとに本番移行を実施しました。最終的な移行実績は以下の通りです。 テーブル規模 テーブル数 LOAD DATA ALTER TABLE 総所要時間 最大規模（数十億レコード） 2 約4日 約7時間 約4日半 中規模（数億レコード） 1 約3時間 約20分 約3時間 小規模（数千万レコード） 5 約1時間 約10分 約1時間 合計 8 - - 約5日 数十時間に及ぶロードでは、以下のクエリで進捗を監視しました。 SET @target_rows = ?; -- 目標件数（テーブルの総行数） SET @thread_id = ?; -- 監視対象のスレッドID SELECT CONCAT ( ' Thread ' , trx.trx_mysql_thread_id) AS target_name, CONVERT_TZ(trx.trx_started, ' UTC ' , ' Asia/Tokyo ' ) AS 開始時刻_JST, ROUND (TIMESTAMPDIFF(SECOND, trx.trx_started, UTC_TIMESTAMP()) / 3600 , 2 ) AS 経過時間 _ 時間, trx.trx_rows_modified AS 挿入済み行数, @target_rows AS 目標件数, ROUND (trx.trx_rows_modified / TIMESTAMPDIFF(SECOND, trx.trx_started, UTC_TIMESTAMP()), 1 ) AS スループット _ 行毎秒, ROUND (trx.trx_rows_modified / @target_rows * 100 , 2 ) AS 進捗率 _ パーセント, ROUND ( (@target_rows - trx.trx_rows_modified) / (trx.trx_rows_modified / TIMESTAMPDIFF(SECOND, trx.trx_started, UTC_TIMESTAMP())) / 3600 , 2 ) AS 残り時間 _ 時間, DATE_ADD( CONVERT_TZ(NOW(), ' UTC ' , ' Asia/Tokyo ' ), INTERVAL ROUND ( (@target_rows - trx.trx_rows_modified) / (trx.trx_rows_modified / TIMESTAMPDIFF(SECOND, trx.trx_started, UTC_TIMESTAMP())) ) SECOND ) AS 完了見込み時刻_JST FROM information_schema.innodb_trx trx WHERE trx.trx_mysql_thread_id = @thread_id; information_schema.innodb_trx の trx_rows_modified から処理済み件数を取得し、経過時間で割ってスループットを算出します。目標件数との差分から残り時間と完了見込み時刻を推定し、数日に及ぶ処理においても見通しを立てられるようにしました。 得られた学び 学び 根拠 並列化は万能ではない マニフェスト分割を試みたが、単一ノードのI/O帯域がボトルネックとなり効果は限定的でした。闇雲に並列化するのではなく、律速段階を特定することが重要です インデックスは後付けが基本 ロード後に一括構築することで最大59％高速化。行挿入ごとのインデックス更新はランダムI/Oを発生させるが、一括構築ならソート後シーケンシャルに処理できる インスタンスサイズはテーブル規模で使い分ける 大規模テーブルはr6i.16xlargeで時間短縮とリスク低減、小規模テーブルはr6i.2xlargeでコスト効率を最大化。スループット向上率とコスト増加率のバランスを見極める 必ず本番同等データでリハーサルする 6,000万レコードでの検証結果を数十億レコードに線形外挿すると誤差が生じる。I/Oやメモリの振る舞いはデータ規模で変化するため、全量リハーサルが不可欠 やり直せる設計が安心を生む S3を中継することでロード失敗時も再抽出不要で再実行できる。数日かかる処理では「失敗しても復旧できる」という安心感が運用の質を高める この工程が安定したことで、後続の増分同期フェーズへ安全に進められました。 Embulkによる差分同期 初期移行が完了した後も、オンプレミスのSQL Serverには新規データが書き込まれ続けます。この増加分をAurora MySQLへ反映するため、 Embulk を用いた差分同期の仕組みを構築しました。 図中の「 マスタ 」はマイクロサービスがSQL Serverをマスタ（書き込みの主系）として参照・更新することを示しています。「 非同期 」はマイクロサービスがSQL Serverと同じ結果をAurora MySQLへ非同期に反映されることを示しています。「 保存 」はEmbulkジョブ完了後に差分の起点となる状態（config-diff）をS3へアップロードすることを指しています。「 復元 」は次回ジョブ起動時にS3からその状態をダウンロードすることを指しています。これにより前回の続きから差分取得を再開できます。 ジョブ設計 Embulkのインクリメンタル同期では、 updated_at のような更新日時カラムを差分キーとして利用するのがベストプラクティスです。しかし、今回の移行元テーブルはInsert/Deleteのみの操作で設計されており、レコードの更新（Update）が発生しないため updated_at に相当するカラムが存在しません。このテーブルの特性を踏まえ、操作種別ごとに差分キーを使い分ける設計を採用しました。 1つのテーブルに対して役割の異なる最大3つのジョブを用意しています。 ジョブ種別 インクリメンタル列 対象レコード 通常ジョブ 登録日（ registered_at ） 新規追加されたレコード 削除ジョブ 削除日（ deleted_at ） 論理削除されたレコード アーカイブジョブ 連番ID 削除テーブルへ移動済みのレコード 通常ジョブは登録日、削除ジョブは削除日をそれぞれ基準にレコードを取得します。 -- 通常ジョブ WHERE registered_at >= :registered_at -- 削除ジョブ WHERE deleted_at IS NOT NULL AND deleted_at >= :deleted_at アーカイブジョブでは、Embulkの before_load と after_load フックを活用し、以下の3ステップを1つのジョブ内で完結させています。 out : mode : merge_direct before_load : > UPDATE watermark SET id = (SELECT COALESCE(MAX(id), 0) FROM archived_favorites) after_load : > DELETE FROM favorites WHERE EXISTS ( SELECT 1 FROM archived_favorites WHERE archived_favorites.favorite_id = favorites.id AND archived_favorites.id >= (SELECT id FROM watermark) ) before_load でロード前のアーカイブテーブルの最大IDをウォーターマークとして記録し、 after_load でウォーターマーク以降の新規アーカイブ分に対応するお気に入りレコードを物理削除します。ウォーターマークがなければアーカイブテーブル全レコードが削除対象となり、毎回全件スキャンが発生します。ウォーターマークにより、今回のジョブで追加された差分だけに処理を限定しています。この設計により、お気に入り商品・ブランド・ショップの各テーブルに対してゲスト・会員の2種類を掛け合わせた複数パターンの差分同期を体系的に管理しています。 ソースDBへの負荷制御 差分同期では稼働中のオンプレミスSQL Serverからデータを読み取ります。本番サービスへの影響を抑えるため、複数のパラメータで負荷を制御しました。 # 共通入力設定（抜粋） in : type : sqlserver transaction_isolation_level : NOLOCK # ロック競合を回避 fetch_rows : 1000 # メモリ消費を抑制 SELECT TOP 10000 -- 1回あたりの取得行数を制限 registered_at, id, member_id, ... FROM favorites WITH (NOLOCK) WHERE registered_at >= :registered_at ORDER BY registered_at OPTION (MAX_GRANT_PERCENT = 25 ) -- クエリのメモリグラント上限を設定 NOLOCK ヒントでロック競合を回避し、 TOP N 句で1回あたりの取得行数を制限しています。 fetch_rows でJDBCのフェッチサイズを制御し、 MAX_GRANT_PERCENT でSQL Serverのクエリメモリグラント上限を設定しました。 また、embulk-input-sqlserverのインクリメンタルロードでは、対応する列型が整数型・文字列型・ datetime2 型に 限定されています 。しかし、移行元テーブルの日時カラムは smalldatetime 型であり、そのままではインクリメンタル列として使用できません。この制約の回避策として、クエリ内で CAST(削除日カラム AS DATETIME) と明示的に型変換しています。 データ整合性の担保 差分取得では > ではなく >= を使用しています。 > の場合、同一タイムスタンプに複数レコードが存在すると一部を取りこぼすリスクがあります。 >= では前回の最終レコードを重複取得する可能性があります。しかし、Embulkの出力モードを merge_direct に設定すれば、重複分はUPSERTとして吸収されます。 out : mode : merge_direct 「取りこぼし」と「重複」のトレードオフにおいて、 重複を許容しつつ冪等性で吸収する 方針を採用しました。 差分の起点となる状態管理にも工夫が必要でした。Embulkは --config-diff オプションにより、前回処理の最終レコード（ last_record ）をYAMLファイルに記録します。 in : last_record : [ '2023-12-23T09:00:30.000000' ] out : {} しかし、Kubernetes Jobとして実行する場合、Podはジョブ完了後に破棄されます。ローカルファイルシステム上の差分状態は失われるため、S3に永続化する仕組みを構築しました。 ジョブ開始時にS3から前回の差分状態をダウンロード Embulkによる差分同期の実行と差分状態の更新 ジョブ完了時に更新された差分状態をS3にアップロード ここで、ダウンロードとアップロードの失敗は致命的エラーとしてジョブを失敗させます。 設定管理とチューニング 複数パターンの設定ファイルは、対象テーブルやカラム名が異なるものの接続情報やパラメータは共通しています。EmbulkのLiquidテンプレート機能を活用し、共通部分を3つのテンプレートに集約しました。 共通テンプレート 役割 入力設定 SQL Server接続情報、トランザクション分離レベル、フェッチサイズ 出力設定 MySQL接続情報、出力モード SELECT句生成 環境変数に基づく TOP N 句の条件付き生成 個別の設定ファイルでは共通テンプレートをインクルードし、テーブル名・カラム名・WHERE句のみを定義します。SELECT句の共通テンプレートでは、環境変数が未設定の場合は TOP 句自体を生成せず、設定されている場合のみ行数制限を付与する条件分岐を実現しています。これにより、本番環境では制限なし、検証環境では制限ありという切り替えが可能です。 負荷制御パラメータ（ TOP N 、 fetch_rows 、 MAX_GRANT_PERCENT 等）もすべて環境変数に切り出しており、コンテナイメージの再ビルドなしに変更を反映できます。テーブル単位で処理時間を計測してボトルネックを特定し、検証環境での調整結果を本番環境へ反映するサイクルを効率的に回せる設計です。 まとめ 本記事では、ZOZOTOWNのお気に入りデータベースにおける数十億レコード規模のデータ移行について、初期移行の最適化とEmbulkを用いた差分同期の取り組みを紹介しました。 初期移行では、インデックスの後付けやテーブル規模に応じたインスタンスサイズの使い分けにより、約5日間で全テーブルの移行を完了しました。差分同期では、 updated_at カラムが存在しない制約に対し、役割の異なる複数ジョブを設計することで、サービス無停止のまま増分データの反映を実現しました。 大規模データ移行やEmbulkによる異種DB間の差分同期を検討されている方にとって、本記事が参考になれば幸いです。今後はAurora MySQL単体運用への切り替えを進め、お気に入り機能のマイクロサービス化を完遂していきます。 ZOZOでは、一緒にサービスを作り上げてくれる方を募集中です。ご興味のある方は、以下のリンクからぜひご応募ください。 corp.zozo.com