Posted inデータベース

MongoDB Change Streamsによるリアルタイムデータ同期の実践ガイド

Database tutorial - IT technology blog
Database tutorial - IT technology blog

ポーリングの代償：なぜ従来の同期手法は大規模環境で破綻するのか

昨年、自社のプロダクトカタログが180万件のドキュメントに達した際、データ同期パイプラインが限界を迎えました。当時は10秒ごとにCeleryワーカーがコレクションをスキャンし、新規の書き込みを検出していました。

# データベースを過負荷に陥らせていた素朴なポーリング手法
last_sync = datetime.utcnow() - timedelta(seconds=10)
updates = db.products.find({"updated_at": {"$gte": last_sync}})

毎分2,000クエリの規模になると、このアプローチは瞬く間に破綻しました。

  • CPUとディスクI/Oの浪費：書き込みが一切発生しない時間帯であっても、インデックススキャンが休むことなく実行されました。同期チェックの負荷だけで、プライマリノードのCPU使用率は45%前後に高止まりしていました。
  • 同期遅延（ラグ）：検索インデックスのデータは、トランザクションストアから最大10秒遅れていました。その結果、すでに在庫切れとなった商品をユーザーがカートに追加してしまうトラブルが日常的に発生しました。
  • 削除の検知漏れ：物理削除（Hard Delete）はログに痕跡を残しません。削除されたドキュメントは$gteクエリから完全に消え去るため、孤立レコードを回収する目的で毎日40分もの照合バッチ処理を実行せざるを得ませんでした。

リレーショナルデータベースではDebeziumやPostgreSQLのレプリケーションスロットのようなWAL（Write-Ahead Log）デコーダーでこの問題を解決しますが、MongoDBはよりスマートな手段を提供しています。それがレプリケーションエンジンに直接組み込まれたイベントキャプチャ機構「Change Streams」です。

Change StreamsがOplogからイベントをキャプチャする仕組み

Change Streamsを使用すると、手動で読み取りクエリを発行することなく、単一のコレクション、データベース全体、あるいはクラスタ全体のドキュメント変更を追跡できます。

内部アーキテクチャの仕組み

各レプリカセットは、oplog.rsという名前の内部cappedコレクションを保持しています。クライアントが書き込みを実行すると、プライマリノードはその操作をこのログに記録し、セカンダリノードへ複製します。Change Streamsはoplogにテイル可能カーソル（tailable cursor）を接続し、生のデータベース操作を構造化された扱いやすいJSONドキュメントへとデコードします。

ストリームを開いておくだけで、過半数のノードへ書き込みがコミットされた瞬間に、MongoDBからネットワーク経由でリアルタイム通知がプッシュされます。

デュアルライト（二重書き込み）より優れている理由

  • 真の原子的整合性：MongoDBとKafkaやRabbitMQなどの外部ブローカーの両方へ書き込むデュアルライトは、スプリットブレインのリスクを孕んでいます。DBコミットとブローカー送信の間でネットワーク障害が起きると、データの不整合が発生します。Change Streamsなら、ストレージへ確実に永続化された変更のみを処理することが保証されます。
  • 耐障害性と復旧性：発行されるすべてのイベントの_idフィールドには、不透明なバイナリ形式の復帰トークン（resume token）が含まれています。予期せぬクラッシュ後にコンシューマーが再起動した場合、再接続時にこのトークンを渡すだけで、ワーカーが中断したまさにその地点から再開できます。
  • ネットワーク転送コストゼロのフィルタリング：$matchや$projectなどの集約ステージをデータベースエンジン内部で直接実行できます。クラスタ側で不要な更新を破棄してから転送するため、ネットワーク帯域を無駄に消費しません。

実践的な実装：堅牢な同期ワーカーの構築

以下は、在庫状況を監視してダウンストリームのサービスと同期を行う、本番環境を見据えた堅牢なPythonワーカーの実装例です。

前提条件

Change Streamsを利用するにはアクティブなoplogが必要なため、レプリカセットが必須となります。ローカルのスタンドアロンインスタンスで開発している場合は、単一ノードのレプリカセットに変換してください。

# レプリカセット名を有効にしてmongodを起動
mongod --port 27017 --dbpath /var/lib/mongodb --replSet rs0

# mongosh内でレプリカセットを初期化
mongosh --eval 'rs.initiate()'

次に、PyMongoをインストールします。

pip install pymongo

1. ストリームの初期化とフィルタリング

コレクション内のすべてのイベントをアプリケーションメモリに流し込むのは避けるべきです。代わりに、集約パイプラインをサーバー側で評価させます。以下のフィルターは、stock（在庫）フィールドが実際に変更されたinsert、update、replace操作にのみマッチします。

import time
import json
from pymongo import MongoClient
from pymongo.errors import PyMongoError

MONGO_URI = "mongodb://localhost:27017/?replicaSet=rs0"
client = MongoClient(MONGO_URI)
db = client["store"]
collection = db["inventory"]

# データベースノード側でイベントをフィルタリング
pipeline = [
    {
        "$match": {
            "operationType": {"$in": ["insert", "update", "replace"]},
            "updateDescription.updatedFields.stock": {"$exists": True}
        }
    }
]

2. 復帰トークン（Resume Token）の管理

再起動に対応できるよう、各イベントの処理が成功するたびに最新の復帰トークンをディスクまたはRedisに書き込みます。

TOKEN_FILE = "last_resume_token.json"

def save_resume_token(token):
    if not token:
        return
    with open(TOKEN_FILE, "w") as f:
        json.dump(token, f)

def load_resume_token():
    try:
        with open(TOKEN_FILE, "r") as f:
            return json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        return None

3. リアクティブなコンシューマーループの実装

トークン管理と再接続ループを組み合わせ、ネットワークの瞬断やクラスタのステップダウンが発生してもリスナーが耐えられるようにします。

def process_change_event(event):
    op_type = event["operationType"]
    doc_id = event["documentKey"]["_id"]
    
    if op_type == "insert":
        product = event.get("fullDocument", {})
        print(f"[同期] アイテムが作成されました: {product.get('name')} (ID: {doc_id})")
    elif op_type in ("update", "replace"):
        updated = event["updateDescription"]["updatedFields"]
        print(f"[同期] 在庫が更新されました (ID: {doc_id}): {updated}")

def run_stream():
    resume_token = load_resume_token()
    
    while True:
        try:
            print("Change Streamのカーソルを開いています...")
            with collection.watch(
                pipeline=pipeline,
                resume_after=resume_token,
                full_document="updateLookup"
            ) as stream:
                for change in stream:
                    process_change_event(change)
                    
                    # 処理が成功した後にのみトークンを確定保存する
                    resume_token = change["_id"]
                    save_resume_token(resume_token)
                    
        except PyMongoError as err:
            print(f"ストリームが中断されました: {err}。5秒後に再試行します...")
            time.sleep(5)

if __name__ == "__main__":
    run_stream()

パイプラインのテスト

1つのターミナルでPythonスクリプトを起動します。別のターミナルからmongoshで接続し、テスト用の新規商品を挿入します。

use store
db.inventory.insertOne({ name: "Mechanical Keyboard", stock: 45 })

わずか15〜25ミリ秒以内に、ワーカーが処理済みイベントを出力します。

[同期] アイテムが作成されました: Mechanical Keyboard (ID: 651fa4bc88a7c1e39...)

本番運用で注意すべき落とし穴

Lookupsによるサーバー負荷の増大

デフォルトでは、updateイベントはupdateDescription内に変更されたキーのみを返します。full_document="updateLookup"を設定すると、MongoDBは変更後のドキュメント全体を取得するために追加クエリを発行します。秒間5,000件の更新が発生する環境では、読み取り負荷が倍増する可能性があります。可能な限り、ドキュメント全体の検索ではなく差分ペイロードを活用してください。

Oplogのローテーション（上書き）による致命的エラー

oplogは固定サイズのリングバッファとして動作します。oplogの保持期間が4時間に設定されている環境でコンシューマーが6時間停止した場合、復帰トークンは無効化されます。この状態になると、サーバーは致命的なChangeStreamHistoryLostエラーをスローします。oplogのサイズは少なくとも24〜48時間分の書き込み量に耐えられるようサイジングし、Prometheusなどでコンシューマーのレプリケーション遅延に対するアラートを設定しておきましょう。

まとめ

ポーリング方式は手軽な反面、遅延、高いCPU負荷、エッジケースの複雑さというトレードオフを抱えています。MongoDB Change Streamsはそのような妥協を排除します。oplogイベントを集約パイプライン経由で直接ストリーミングすることで、わずかなコード量でサブセカンド単位の高速な同期と堅牢なクラッシュ復旧を実現できます。

Share:
Tags: