Amazon DocumentDB の Change Streams で構築するイベント駆動アーキテクチャ
DocumentDB の変更データキャプチャで Lambda トリガーと連携し、リアルタイムデータ同期のイベント駆動アーキテクチャを構築する手法を紹介します。
Change Streams の仕組み
DocumentDB の Change Streams はコレクションに対する変更 (挿入、更新、削除、置換) をリアルタイムにキャプチャする機能です。MongoDB の Change Streams API と互換性があり、既存の MongoDB アプリケーションのコードをそのまま使用できます。Change Streams はクラスタの変更ログ (oplog) を基盤としており、変更が発生した順序で配信されます。各イベントには operationType (insert、update、delete)、documentKey (変更されたドキュメントの ID)、fullDocument (変更後のドキュメント全体) が含まれます。
Lambda トリガーとの統合
DocumentDB の Change Streams を Lambda のイベントソースマッピングとして設定すると、変更イベントが自動的に Lambda 関数に配信されます。Lambda 関数内でイベントの種類に応じた処理を実行します。典型的なユースケースは、OpenSearch のインデックス更新 (ドキュメントの変更を検索インデックスに即座に反映)、ElastiCache のキャッシュ無効化 (変更されたドキュメントのキャッシュを削除)、DynamoDB へのデータ同期 (読み取り最適化されたビューの維持)、SNS 通知 (特定の条件に合致する変更の通知) です。バッチサイズとバッチウィンドウを設定し、複数のイベントをまとめて処理することでスループットを向上できます。
- 1. 変更が oplog に記録されるコレクションへの挿入・更新・削除・置換がクラスタの変更ログ (oplog) に書かれる。Change Streams はこの oplog を基盤にしており、変更が発生した順序で配信される。
- 2. 変更イベントが組み立てられる各イベントには operationType (insert、update、delete)、documentKey (変更されたドキュメントの ID)、fullDocument (変更後のドキュメント全体) が入る。MongoDB の Change Streams API と互換なので、既存のアプリケーションのコードをそのまま使える。
- 3. Lambda のイベントソースマッピングへ配信されるイベントソースマッピングとして設定しておくと、変更イベントが自動的に Lambda 関数へ渡る。バッチサイズとバッチウィンドウで複数イベントをまとめると、スループットを上げられる。
- 4. 下流へ反映し、resume token で再開点を持つOpenSearch のインデックス更新、ElastiCache のキャッシュ無効化、DynamoDB への同期、SNS 通知などを実行する。各イベントに含まれる resume token を押さえておけば、障害復旧時に最後に処理したイベントの次から読み直せる (イベントソースマッピングでは自動)。
障害復旧と運用上の注意点
Change Streams の各イベントには resume token が含まれており、この token を保存しておくことで、障害復旧時に最後に処理したイベントの次から読み取りを再開できます。 Lambda のイベントソースマッピングでは resume token の管理が自動化されています。 Change Streams の変更ログは最大 7 日間保持されるため、 7 日以上処理が停止すると古いイベントが失われます。長時間の障害に備えて、 Change Streams とは別に定期的なフルスナップショットを取得する設計が推奨されます。 Change Streams の有効化はクラスタのパフォーマンスに若干の影響を与えるため、本番環境では事前にパフォーマンステストを実施します。
Change Streams の料金考慮
Change Streams 自体に追加料金は発生しませんが、変更ログの保持に I/O とストレージが消費されます。変更ログの保持期間は最大 7 日間で、書き込みが多いコレクションではストレージ使用量が増加します。Lambda トリガーを使用する場合、Lambda の呼び出し回数と実行時間の料金が発生します。変更頻度が高いコレクションでは、Lambda のバッチサイズを大きく設定して呼び出し回数を削減し、コストを最適化します。Kinesis Data Streams への転送を選択した場合は、シャード時間の料金が追加されます。
変更データキャプチャの活用範囲
Change Streams は、データの変更を起点にさまざまな処理を連動させる土台になります。代表的なのは、検索エンジンのインデックスをドキュメントの変更に追従させる用途や、変更されたデータのキャッシュを無効化して常に最新を返す用途です。読み取りに最適化した別のビューを維持する、変更を分析基盤へ流して集計する、といった使い方も広がります。書き込み側のアプリケーションに手を入れずに、後段の処理を疎結合に追加できる点が利点で、システムを段階的に拡張していく際の柔軟性を高めます。
変更イベントのフィルタリングと整形
Change Streams は MongoDB 互換の集約パイプラインを使い、受け取る変更を絞り込めます。特定のコレクションや操作種別だけを対象にする、関心のあるフィールドが変わったときだけ反応する、といった制御が可能です。更新イベントで変更後のドキュメント全体を取得するオプションを使えば、差分だけでなく完全な状態を後続処理へ渡せます。必要なイベントだけを通すことで、下流の処理負荷とコストを抑えられます。受け取り側で全件をふるいにかけるのではなく、ソースに近い段階で絞り込む設計が効率的です。
Kinesis や MSK へのファンアウト
変更イベントを複数の処理系へ届けたい場合は、Lambda を介して Kinesis Data Streams や MSK へ転送する構成が有効です。ストリーミング基盤に一度集約すれば、複数のコンシューマーがそれぞれの目的で同じ変更を消費でき、検索更新・分析・通知などを並行して走らせられます。順序が重要なデータは、キーの設計で同一シャードやパーティションに集約して順序を保ちます。直接 Lambda で処理する単純な構成から、ストリーミング基盤を挟む拡張性の高い構成へと、要件の成長に応じて発展させられます。
設計上の注意とベストプラクティス
Change Streams を使ったイベント駆動では、いくつかの定石を押さえます。同じイベントが再配信されても問題が起きないよう、処理は冪等に設計します。レジュームトークンを使って中断箇所から再開できるようにし、変更ログの保持期間を超えて処理が止まると古い変更が失われる点に備えて、別途スナップショットも取得します。有効化はクラスタの負荷に影響しうるため、本番投入前に性能を検証します。遅延や失敗を監視し、滞留が増えたら早めに対処することで、信頼性の高い変更連携を維持できます。
まとめ
DocumentDB の Change Streams はドキュメントの変更をリアルタイムにキャプチャし、イベント駆動アーキテクチャを構築する機能です。Lambda トリガーで変更を自動処理し、Kinesis Data Streams への統合でダウンストリームの分析パイプラインにデータを配信します。レジュームトークンで処理の再開位置を管理し、信頼性の高い変更データキャプチャを実現します。
参考資料 (AWS 公式)
本ページの一次情報は AWS 公式サイトおよび公式ドキュメントです。最新の仕様 / 料金は次の公式ページで確認できます。
本ページと公式ドキュメントの記述が食い違う場合は、公式ドキュメントを正としてください。