Production practice of Flink MongoDB CDC in XTransfer

まえがき

XTransfer は、越境 B2B E コマースの中小企業向けに、越境金融およびリスク管理サービスを提供することに注力しています。現地受取口座、外国為替、海外為替管理国申告など、さまざまな越境金融サービスに対する包括的なソリューションを提供しています。

ビジネス開発の初期段階では、従来のオフラインデータウェアハウスアーキテクチャを採用し、フル収集、バッチ処理、上書き書き込みによる Data Integration 方式を利用していました。しかし、データの即時性に課題がありました。ビジネスの発展に伴い、オフラインデータウェアハウスではデータの即時性に関する要件を満たすことが難しくなったため、オフラインデータウェアハウスからリアルタイムデータウェアハウスへの進化を決定しました。リアルタイムデータウェアハウス構築の鍵は、データ収集ツールとリアルタイムコンピューティングエンジンの選定にあります。

一連の調査を経て、2021 年 2 月に Flink CDC プロジェクトに注目しました。Flink CDC には Debezium が組み込まれており、Flink 自体が変更データキャプチャ機能を備えています。これにより、開発のハードルが大幅に下がり、デプロイの複雑さも軽減されます。Flink の強力なリアルタイムコンピューティング機能と豊富な外部システム接続機能により、リアルタイムデータウェアハウス構築の重要なツールとなりました。

さらに、本番環境で MongoDB を広く活用しているため、Flink CDC を基盤として MongoDB の Change Streams 機能を活用し、Flink MongoDB CDC Connector を実装して Flink CDC コミュニティに貢献しました。バージョン 2.1 でリリースされています。ここで、その実装の詳細と本番環境での実践を共有できることを大変光栄に思います。

1. Flink CDC

動的テーブル (Dynamic Table) は、ストリーミングデータをサポートする Flink の Table API および SQL の中核概念です。ストリームとテーブルには双対性があり、テーブルを changelog ストリームに変換することも、changelog ストリームを再生してテーブルに復元することもできます。

変更ストリームには、Append Mode と Update Mode の 2 種類があります。Append Mode は追加のみで変更や削除は行いません。イベントフローなどがこれに該当します。Update Mode は追加、変更、削除が可能です。データベースの操作ログなどがこれに該当します。Flink 1.11 より前は、Append Mode の動的テーブルのみがサポートされていました。

Flink 1.11 では、FLIP-95 で新しい TableSource と TableSink が導入され、Update Mode の changelog がサポートされました。また、FLIP-105 では、Debezium および Canal CDC フォーマットの直接サポートが導入されました。ScanTableSource を実装して外部システムの変更ログ (データベースの変更ログなど) を受信し、Flink が認識できる changelog として解釈して下流に流すことで、変更ログからの動的テーブル定義をサポートできます。

Flink 内部では、changelog レコードは RowData で表現され、+I (INSERT)、-U (UPDATE_BEFORE)、+U (UPDATE_AFTER)、-D (DELETE) の 4 種類があります。changelog が生成するレコードタイプの違いにより、3 つの changelog モードに分類できます。


2. MongoDB のレプリケーションメカニズム

前節で述べたように、Flink CDC MongoDB を実装する際の鍵は、MongoDB の操作ログを Flink がサポートする changelog にどのように変換するかです。この問題を解決するには、まず MongoDB のクラスターデプロイとレプリケーションメカニズムを理解する必要があります。

2.1 レプリカセットとシャードクラスター
レプリカセットは、MongoDB が提供する高可用性デプロイモードです。レプリカセットのメンバーは oplog (操作ログ) をレプリケーションすることで、レプリカセットメンバー間のデータ同期を実現します。

シャードクラスターは、MongoDB が大規模データセットと高スループット操作をサポートするためのデプロイモードで、各シャードはレプリカセットで構成されます。

2.2 レプリカセットの oplog

操作ログ oplog は、MongoDB において特別な Capped コレクション (固定容量コレクション) で、データの操作ログを記録し、レプリカセットメンバー間の同期に使用されます。oplog レコードのデータ構造は以下の通りです。

例から分かるように、MongoDB oplog の更新レコードには、更新前の情報も更新後の完全なレコードも含まれていません。そのため、Flink がサポートする ALL タイプの changelog に変換することも、UPSERT タイプの changelog に変換することもできません。

さらに、シャードクラスターでは、データの書き込みが異なるシャードのレプリカセットで発生する可能性があるため、各シャードの oplog にはそのシャードで発生したデータ変更のみが記録されます。したがって、完全なデータ変更を取得するには、各シャードの oplog を操作時刻の順序に従ってマージする必要があり、変更レコードのキャプチャの難易度とリスクが高まります。

Debezium MongoDB Connector はバージョン 1.7 以前、oplog を走査することで変更データキャプチャを実現していました。上記の理由から、Debezium MongoDB Connector ではなく、MongoDB 公式の Change Streams ベースの MongoDB Kafka Connector を選択しました。

2.3 Change Streams

Change Streams は MongoDB 3.6 で導入された新機能で、oplog の走査の複雑さを隠蔽し、シンプルな API を通じてクラスター、データベース、コレクションレベルでデータ変更をサブスクライブできます。

2.3.1 使用条件
2.3.2 Change Event
2.3.3 Update Lookup

oplog の更新操作には変更されたフィールドのみが含まれるため、oplog から直接変更後の完全なドキュメントを取得できません。しかし、UPSERT モードの changelog に変換する際、UPDATE_AFTER の RowData には完全な行レコードが必要です。Change Streams は fullDocument = updateLookup を設定することで、変更レコードの取得時にドキュメントの最新状態を返せます。さらに、各 Change Event レコードには documentKey (_id とシャードキー) が含まれており、変更されたレコードのプライマリキー情報を識別できます。これはべき等更新の条件を満たします。したがって、Update Lookup 機能を通じて、MongoDB の変更レコードを Flink の UPSERT changelog に変換できます。

3. Flink MongoDB CDC

具体的な実装としては、Change Streams ベースで実装された MongoDB 公式の MongoDB Kafka Connector を統合しました。Debezium EmbeddedEngine を使用することで、MongoDB Kafka Connector を Flink 内で簡単に実行できます。MongoDB CDC TableSource は、Change Stream を Flink UPSERT changelog に変換することで実装されています。Change Streams の resume メカニズムを活用して、チェックポイントおよびセーブポイントからの復元機能を実現しています。

FLIP-149 に記載されているように、一部の操作 (集計など) は -U メッセージなしでは正しく処理することが困難です。UPSERT タイプの changelog に対して、Flink Planner は追加のコンピューティングノード (Changelog Normalize) を導入し、ALL タイプの changelog に正規化します。


サポート機能

Exactly-Once セマンティクスのサポート
フルおよび増分サブスクリプションのサポート
スナップショットデータフィルタリングのサポート
チェックポイント、セーブポイントからの復元のサポート
メタデータ抽出のサポート

4. 本番環境での実践

4.1 RocksDB State Backend の使用

Changelog Normalize は、-U の前置镜像値を補完するために追加のステートオーバーヘッドをもたらします。本番環境では RocksDB State Backend の使用を推奨します。

4.2 適切な oplog 容量と保持時間

MongoDB の oplog.rs は容量を持つ特別なコレクションです。oplog.rs の容量が最大値に達すると、過去のデータは破棄されます。Change Streams は resume token を通じて復元されますが、oplog 容量が小さすぎると、resume token に対応する oplog レコードが存在しなくなり、復元に失敗する可能性があります。

oplog 容量が明示的に指定されていない場合、WiredTiger エンジンのデフォルト oplog 容量はディスクサイズの 5% で、下限は 990 MB、上限は 50 GB です。MongoDB 4.4 以降では、oplog の最小保持時間の設定がサポートされています。oplog が一杯になり、かつ oplog レコードが最小保持時間を超えた場合、その oplog レコードは回収されます。

oplog 容量と最小保持時間は replSetResizeOplog コマンドで再設定できます。本番環境では、oplog 容量を 20 GB 以上、oplog 保持時間を 7 日以上にするよう推奨します。

4.3 変更の少ないコレクションでのハートビートイベントの有効化

Flink MongoDB CDC は定期的に resume token をチェックポイントに書き込み、Change Stream の復元を行います。MongoDB の変更イベントまたはハートビートイベントが resume token の更新をトリガーします。サブスクライブしているコレクションの変更が少ない場合、最後の変更レコードに対応する resume token が期限切れとなり、チェックポイントからの復元ができなくなる可能性があります。したがって、変更の少ないコレクションでは、ハートビートイベントを有効化 (heartbeat.interval.ms > 0 に設定) して、resume token を常に最新に保つことを推奨します。

4.4 MongoDB 接続パラメータのカスタマイズ

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

4.5 Change Stream パラメータのチューニング
設定変更イベントのポーリングは、Flink DDL の poll.await.time.ms と poll.max.batch.size で細かく調整できます。

poll.await.time.ms
変更イベントのポーリング時間間隔で、デフォルトは 1,500 ms です。変更頻度の高いコレクションでは、ポーリング間隔を適切に短縮して処理効率を向上できます。変更の少ないコレクションでは、ポーリング間隔を適切に延長してデータベースへの負荷を軽減できます。

poll.max.batch.size
各バッチでポーリングする変更イベントの最大数で、デフォルトは 1,000 です。このパラメータを増加させると、Cursor から変更イベントをポーリングする速度が向上しますが、メモリオーバーヘッドが増加します。

4.6 データベース全体およびクラスター全体の変更のサブスクライブ
database = "db"、collection = "" と設定すると、db データベース全体の変更をサブスクライブできます。database = ""、collection = "" と設定すると、クラスター全体の変更をサブスクライブできます。

DataStream API では、パイプラインを使用してサブスクライブ対象のデータベースとコレクションをフィルタリングできます。スナップショットコレクションのフィルタリングは現時点ではサポートされていません。

4.7 アクセス制御
MongoDB は、ユーザー、ロール、権限のきめ細かな制御をサポートしています。Change Stream を有効にするユーザーには、find と changeStream の 2 つの権限が必要です。

単一コレクション

クラスター

{ resource: { db: "", collection: "" }, actions: [ "find", "changeStream" ] }
本番環境では、Flink 専用のユーザーとロールを作成し、ロールに対してきめ細かな権限付与を行うことを推奨します。MongoDB では任意のデータベース配下でユーザーとロールを作成できる点に注意が必要です。ユーザーが admin 配下で作成されていない場合、接続パラメータに authSource = <ユーザーが存在するデータベース> を指定する必要があります。

開発環境およびテスト環境では、read と readAnyDatabase の 2 つの組み込みロールを Flink ユーザーに付与することで、任意のコレクションの Change Stream を有効化できます。