Implementation Principle and Practice of Flink CDC MongoDB Connector

1. MongoDB Change Streams テクノロジーの概要

MongoDB は、半構造化データストレージをサポートするドキュメント指向の非リレーショナルデータベースです。また、レプリカセットとシャードクラスターの 2 つのクラスターデプロイモードを提供する分散データベースでもあり、高可用性と水平スケーラビリティを備えています。大規模データストレージに適しています。さらに、MongoDB 4.0 ではマルチドキュメントトランザクションのサポートも追加され、より複雑なビジネスシナリオにも対応しやすくなりました。

MongoDB は弱構造ストレージモデルを採用しており、柔軟なデータ構造と豊富なデータ型をサポートしています。JSON ドキュメント、タグ、スナップショット、地理位置情報、コンテンツストレージなどのビジネスシナリオに適しています。ネイティブな分散アーキテクチャにより、すぐに使えるシャーディングメカニズムと自動リバランス機能を提供し、大規模データストレージに対応しています。また、MongoDB には GridFS という分散グリッドファイルストレージ機能があり、画像、音声、動画などの大容量ファイルの保存に適しています。

MongoDB はレプリカセットとシャードクラスターの 2 つのクラスターデプロイモードをサポートしています。

レプリカセット：高可用性を実現するデプロイモードで、セカンダリノードはプライマリノードの操作ログを複製することでデータを同期します。プライマリノードに障害が発生した場合、セカンダリノードとアービターノードが投票を再実施して新しいプライマリノードを選出し、フェールオーバーを実現します。また、セカンダリノードはクエリリクエストを分担し、プライマリノードのクエリ負荷を軽減することもできます。

シャードクラスター：水平スケールを実現するデプロイモードで、データを異なるシャードに均等に分散させます。各シャードはレプリカセットとしてデプロイできます。シャード内のプライマリノードが読み書きリクエストを処理し、セカンダリノードはプライマリノードの操作ログを複製します。指定されたシャーディングキーとシャーディングポリシーに基づき、データは複数の 16 MB のデータチャンクに分割され、異なるシャードに格納されます。シャードとデータチャンクの対応関係は Config Server に記録されます。

MongoDB の oplog は MySQL のバイナリログに似ており、MongoDB 内のデータに対するすべての操作ログを記録します。oplog は容量制限付きのコレクションで、設定された容量を超えると古い情報が破棄されます。

MySQL のバイナリログとは異なり、oplog は変更前後の完全な情報を記録しません。oplog を走査することで MongoDB のデータ変更をキャプチャできますが、Flink がサポートするチェンジログへの変換にはいくつかの制約があります。

まず、oplog のサブスクライブは複雑です。各レプリカセットは独自の oplog を管理しています。シャーディングされたクラスターでは、各シャードが独立したレプリカセットである可能性があり、各シャードの oplog を走査して操作時間順にソートする必要があります。さらに、oplog には変更ドキュメントの前後の完全な状態が含まれていないため、Flink 標準のチェンジログにも Upsert チェンジログにも変換できません。これが、MongoDB CDC Connector の実装において oplog を直接サブスクライブする方式を採用しなかった主な理由です。

最終的に、MongoDB 3.6 で提供された Change Streams を使用して MongoDB CDC Connector を実装することを選択しました。

Change Streams は MongoDB 3.6 で導入された新機能で、よりシンプルな変更データキャプチャインターフェイスを提供し、oplog を直接走査する複雑さを隠蔽します。Change Streams は変更ドキュメントの完全な状態の抽出機能も提供し、Flink の Upsert 型チェンジログに簡単に変換できます。また、比較的完全な障害復旧機能も備えており、各変更レコードには現在の Change Streams の位置を記録するレジュームトークンが含まれています。障害発生後、レジュームトークンを使用して現在の消費ポイントから再開できます。

さらに、Change Streams は変更イベントのフィルタリングとカスタマイズをサポートします。たとえば、データベース名とコレクション名の正規表現フィルターを MongoDB 側にプッシュダウンして実行できるため、ネットワークオーバーヘッドを大幅に削減できます。コレクション、データベース、およびクラスター全体レベルでの変更サブスクリプションを提供し、対応する権限制御もサポートします。

MongoDB Change Streams 機能を使用して実装された CDC Connector は上図の通りです。まず、Change Streams を通じて MongoDB の変更をサブスクライブします。たとえば、insert、update、delete、replace の 4 種類の変更があります。まず、Flink がサポートする upsert チェンジログに変換し、その上に動的テーブルを定義して Flink SQL で処理します。

現在、MongoDB CDC Connector は Exactly-Once セマンティクスをサポートし、フルデータと増分の両方を含むサブスクリプション、チェックポイントとセーブポイントからのリカバリ、スナップショットデータのフィルタリング、データベースやコレクションなどのメタデータ抽出、コレクションの正規表現フィルタリング機能をサポートしています。

2. MongoDB CDC Connector のビジネス実践

2017 年に設立された XTransfer は、B2B 国境間決済事業に注力し、国境間越境 EC 輸出に従事する中小零細企業向けに貿易集金とリスク管理サービスを提供しています。B2B 国境間決済シナリオにおけるビジネスフローは非常に長く、照会から最終取引までの過程で物流条件、決済条件などが関わり、各環節で国境間資金取引の規制要件に準拠するためのリスク管理が必要です。

上記の要素から、XTransfer のデータ処理には高いセキュリティと精度が求められます。この基盤の上で、XTransfer は Flink ベースのビッグデータプラットフォームを構築し、B2B フルリンク上のデータを効果的に収集・処理・計算でき、高いセキュリティ、低レイテンシ、高精度の要件を満たしています。

変更データキャプチャ（CDC）はデータ統合の重要な部分です。Flink CDC を使用する前は、一般的に Debezium や Canal などの従来の CDC ツールを使用してデータベースの変更ログを抽出して Kafka に転送し、下流で Kafka の変更ログを読み取って消費していました。このアーキテクチャには以下の課題があります。

・デプロイコンポーネントが多く、運用保守コストが高い。
・下流のデータ消費ロジックを書き込み元に合わせて適応する必要があり、一定の開発コストが発生する。
・データサブスクリプションの設定が煩雑で、Flink CDC のように SQL 文のみで完全なデータ同期ロジックを定義できない。
・フルデータと増分データの収集を完全に満たすことが難しく、DataX などのフルデータ収集コンポーネントを導入する必要がある場合がある。
・変更データの収集に偏っており、データ処理やフィルタリング機能が比較的弱い。
・異種データソースの拡幅シナリオに対応しにくい。

現在、当社のビッグデータプラットフォームは主に Flink CDC を変更データキャプチャに使用しており、以下の利点があります。

1. リアルタイムデータ統合

Debezium、Canal、DataX などの追加コンポーネントをデプロイする必要がなく、運用保守コストが大幅に削減されます。
豊富なデータソースをサポートし、Flink の既存のコネクタを再利用してデータの収集と書き込みができ、ほとんどのビジネスシナリオをカバーできます。
開発の難易度を下げ、Flink SQL のみで完全なデータ統合ワークフローを定義できます。
強力なデータ処理能力を備え、Flink プラットフォームの強力な演算能力により、ストリーミング ETL や異種データソースの結合、集計を実現できます。

2. リアルタイムデータウェアハウスの構築

リアルタイムデータウェアハウスのデプロイ難易度を大幅に簡素化し、Flink CDC でデータベースの変更をリアルタイムに収集して Kafka、Iceberg、Hudi、TiDB などのデータベースに書き込み、その後 Flink で深層データマイニングとデータ処理を実行できます。
Flink の計算エンジンはストリーミングとバッチの統合計算モードをサポートしています。複数の計算エンジンを保守する必要がなくなり、データ開発コストを大幅に削減できます。

3. リアルタイムリスク管理

従来、リアルタイムリスク管理は通常ビジネスイベントを Kafka に送信して実装されていました。Flink CDC を使用すると、ビジネスデータベースから直接リスク管理イベントをキャプチャし、Flink CDC で複雑なイベント処理を実行できます。
Flink ML や Alink を通じてモデルを実行し、機械学習機能を強化できます。最終的に、これらのリアルタイムリスク管理の処分結果を Kafka に送信し、リスク管理指令を発行します。

3. MongoDB CDC Connector の本番環境チューニング

MongoDB CDC Connector の使用には以下の要件があります。

Change Streams 機能を使用して MongoDB CDC Connector を実装しているため、MongoDB の最低利用可能バージョンは 3.6 以上で、4.0.8 以上を推奨します。
クラスターデプロイモードを使用する必要があります。MongoDB の Change Streams のサブスクライブにはノード間のデータレプリケーションが必要ですが、単一の MongoDB ではノード間のデータレプリケーションが行われず、oplog も存在しません。レプリカセットまたはシャードクラスターにのみデータレプリケーションメカニズムがあります。
WireTiger ストレージエンジンを使用し、pv1 レプリケーションプロトコルを使用する必要があります。
ChangeStream および find 権限が必要です。

MongoDB CDC Connector の使用時には、oplog の容量と有効期限の設定に注意してください。MongoDB の oplog は容量制限付きの特殊なコレクションで、容量が最大値に達すると履歴データが破棄されます。ただし、Change Streams はレジュームトークンを通じて復旧します。oplog 容量が小さすぎると、レジュームトークンに対応する oplog レコードが存在しなくなる可能性があり、つまりレジュームトークンが失効し、Change Streams が復旧できなくなります。

replSetResizeOplog を使用して oplog の容量と最小保持時間を設定できます。MongoDB 4.4 以降では最小時間の設定もサポートされています。一般的に、本番環境では oplog を 7 日以上保持することを推奨します。

変更頻度の低いテーブルでは、設定でハートビートイベントを有効にすることを推奨します。変更イベントとハートビートイベントは同時にレジュームトークンを進行できます。変更頻度の低いテーブルでは、ハートビートイベントを通じてレジュームトークンを更新し、その期限切れを回避できます。

ハートビート間隔は heartbeat.interval.ms で設定できます。

MongoDB の Change Streams は Flink の Upsert チェンジログにしか変換できないため、Upsert Kafka と同様に、-U の前イメージ値を補完する ChangelogNormalize オペレーターが追加され、追加の状態オーバーヘッドが発生します。そのため、本番環境では RocksDB State Backend の使用を推奨します。

デフォルトの接続パラメーターで要件を満たせない場合は、connection.options 設定項目を通じて MongoDB がサポートする接続パラメーターを渡すことができます。

たとえば、MongoDB に接続するユーザーが作成したデータベースが admin でない場合、パラメーターを設定して現在のユーザーの認証に使用するデータベースを指定でき、接続プールの最大接続パラメーターも設定できます。MongoDB の接続文字列はデフォルトでこれらのパラメーターをサポートしています。

複数データベースと複数テーブルの正規表現マッチングは、MongoDB CDC Connector バージョン 2.0 以降で提供される新機能です。注意すべき点として、データベース名に正規表現パラメーターを使用する場合、readAnyDatabase ロールが必要です。MongoDB の Change Streams はクラスター全体、データベース、コレクションの粒度でしか有効化できないためです。データベース全体をフィルタリングする必要がある場合、データベースに正規表現を使用する際にクラスター全体で Change Streams を有効化してから、Pipeline を通じてデータベースの変更をフィルタリングするしかありません。データベースとコレクションの 2 つのパラメーターに正規表現を記述することで、複数のデータベースとテーブルをサブスクライブできます。

4. MongoDB CDC Connector の並列スナップショット改善

スナップショットの速度を向上させるため、Flip-27 で導入された source を使用した並列化変換が可能です。まず、split enumerator が完全なスナップショットタスクを特定の分割戦略に従って複数のサブタスクに分割し、その後複数の split reader に割り当てて並列でスナップショットを実行することで、全体のタスク実行速度を向上させます。

ただし MongoDB では、ほとんどの場合コンポーネントは ObjectID で、最初の 4 バイトは UNIX タイムスタンプ、中間の 5 バイトはランダム値、最後の 3 バイトは自動インクリメントです。同じタイムスタンプで挿入されたドキュメントは厳密には増加せず、中間のランダム値が局所的な厳密増加に影響する可能性がありますが、全体的には増加傾向を満たしています。

そのため、MySQL の増分コンポーネントとは異なり、MongoDB のコレクションに offset + limit 分割戦略を単純に適用するのは適しておらず、ObjectID に特化した分割戦略が必要です。

最終的に、以下の 3 つの MongoDB 分割戦略を採用しました。

サンプルサンプリングバケット化：$sample コマンドを使用してコレクションをランダムサンプリングし、平均ドキュメントサイズと各チャンクのサイズから必要なバケット数を推定する方式です。対応コレクションのクエリ権限が必要です。処理速度が速く、大量データを持つ非シャーディングコレクションに適しているのが利点です。サンプリング推定モードを使用するため、バケット化結果が完全に均一にならないのが欠点です。

SplitVector インデックス分割：SplitVector は MongoDB がチャンクの分割点を計算する内部コマンドで、指定されたインデックスにアクセスして各チャンクの境界を算出します。SplitVector 権限が必要です。処理速度が速くチャンク結果が均一なのが利点です。大量データを持つシャーディング済みコレクションでは、config ライブラリから既存のチャンクメタデータを直接読み取る方が効率的なのが欠点です。

Chunks メタデータ読み取り：MongoDB はシャーディングコレクションの実際のシャーディング結果を config データベースに保存しているため、config から直接そのシャーディング結果を読み取れます。config ライブラリへの読み取りアクセス権限が必要で、シャーディングコレクションにのみ適用可能です。チャンク分割点を再計算する必要がなく、チャンク結果は均一でデフォルトサイズは 64 MB という利点があります。すべてのシナリオに対応できるわけではなく、シャーディングシナリオに限定されるのが欠点です。

上図はサンプルサンプリングバケット化の例です。左側は完全なコレクションです。完全なコレクションからサンプル数を設定し、サンプル全体を縮小してから、サンプリングされたサンプルに基づいてバケット化を実行します。最終的な結果が求めるチャンク境界です。

sample コマンドは MongoDB のサンプリング用組み込みコマンドです。サンプル値が 5% 未満の場合、疑似ランダムアルゴリズムを使用してサンプリングを行います。サンプル値が 5% の場合、まずランダムソートを行い、上位 N 件のドキュメントを選択します。均一性と処理時間は主にランダムアルゴリズムとサンプル数に依存します。均一性と分割速度のトレードオフを取る戦略で、高速な分割速度が必要だが分割結果の不均一さを許容できるシナリオに適しています。

実際のテストでは、サンプルサンプリングの均一性は良好なパフォーマンスを示しています。

上図は SplitVector インデックス分割の例です。左側は元のコレクションで、SplitVector コマンドでアクセスするインデックスを指定します。ここでは ID インデックスです。各チャンクのサイズを MB で設定し、SplitVector コマンドでインデックスにアクセスして、インデックスを通じて各チャンクの境界を計算できます。

処理速度が速くチャンク結果が非常に均一で、ほとんどのシナリオに適しています。

上図は config.chunks の読み取り例で、MongoDB が既に分割したチャンクのメタデータを直接読み取ります。各シャードとそのマシン、各シャードの境界は Config Server に保存されています。シャーディングコレクションでは、chunks から境界情報を直接読み取れるため、分割点を繰り返し計算する必要がなく、各チャンクの読み取りが単一マシンで完了するため極めて高速で、大規模なシャーディングコレクションシナリオで優れたパフォーマンスを発揮します。

5. 今後の計画

Flink CDC の今後の計画は主に以下の 5 つの側面に分かれます。

1 つ目、Flink CDC 増分スナップショットフレームワークの改善を支援する。
2 つ目、MongoDB CDC を Flink CDC 増分スナップショットフレームワークに接続し、並列スナップショット改善をサポートできるようにする。
3 つ目、MongoDB CDC で Flink RawType をサポートする。より柔軟なストレージ構造に対して RawType 変換を提供し、ユーザーは UDF 形式でカスタム解析を実行できる。
4 つ目、MongoDB CDC で指定位置からの変更データ収集をサポートする。
5 つ目、MongoDB CDC の安定性最適化。