調査_OpenSerachデータ同期の仕組み検討（DMS/Glue）
背景
現状のデータ同期の仕組みではデータ全同期/差分同期に時間がかかり過ぎている。
データ仕組みの方法を見直し、手軽にindex作成、同期が行える環境を構築したい。
選択肢としてあるのが「DMS」「Glue」を用いた方法である。それぞれについて実現方法を調査する。
DMSの場合
複雑な事前処理が難しい。
複雑なJOINクエリをそのまま実行することは出来ないため、データ同期用のテーブルを別途用意する。
前提、ES側ではJOINが不可。
アプリ側で「ユーザー情報、部署情報、権限情報を取得したい」場合は以下のような方法が必要。
A. Aurora側に users_with_dept のような非正規化テーブルを作っておき、それをDMSによって1レコード = 1ドキュメント として流す。
B. 絞り込みはユーザーのみに行い、他情報はRDBから取得
Aurora側に非正規化テーブルを用意する→データの整合性を取るのが面倒。複雑になる。
アプリ側のクエリ修正も必要になる
→ あまり現実的では無い
Glue
-
全データ同期
「Auroraから全件取得。Spark(PySpark)で変換。OSにBulk投入」のようなジョブを作成。
時間がかかる部分はDPUの増加で可能。 -
差分同期
DMSのような「ログベース CDC」の仕組みは無い。
「最終同期時間を持っておき、それ以降のデータ更新分を取得＆変換＆反映する」といった方法が取れる。
まとめると
Glue：
フル同期・・・そのまま全件読みで可能
差分同期・・・「updated_at ベース」など、アプリ側のルールで実現
DMS：
フル同期・・・設定で「フルロード」
差分同期・・・その後の CDC（ログベース）で標準サポート
DMSの方が標準機能でデータ同期をサポートしてくれているが、変換が複雑になると難しい。
今回はGlueの方が良さそう。
用語解説
ETL（Extract / Transform / Load）
データ連携のこと。取得/変換/投入。
E：Extract（抽出）
Aurora（RDB）などのソースからデータを取ってくる部分
例：SELECT * FROM users WHERE updated_at >= :last_synced_at
T：Transform（変換）
取得した行を、「使いたい形」に変換する部分
例：
・複数テーブルの JOIN
・カラム名の変更
・JSON を組み立てる
・検索用にテキスト結合する など
L：Load（書き込み）
変換済みのデータを、ターゲット（OpenSearch など）に投入する部分
例：Bulk API でインデックスにドキュメントを投入
CDC（Change Data Capture）
変更されたデータだけを検知して同期する仕組みのこと。差分同期。
ログベース CDC
DB のトランザクションログ（binlog / WAL）を読んでクエリをそのまま流す方式。
DMS が基本的にやっているのはこれ。
タイムスタンプベース CDC
テーブルに updated_at を持たせて
WHERE updated_at >= :last_sync_time のように差分を取る方式
トリガーベース CDC
DB のトリガーで「更新があったら専用テーブルに書く」「イベントを飛ばす」方式
現状は「タイムスタンプベースCDC」に近い
Discussion