ClickHouse transaction implementation of EMR OLAP open source real-time data warehouse solution

1、背景

Flink と ClickHouse は、それぞれリアルタイムストリームコンピューティングと OLAP のリーダーです。インターネット、広告、ゲームなどの多くのお客様が、この 2 つを組み合わせてユーザープロファイル、リアルタイム BI レポート、アプリケーションモニタリングのインデックスクエリ、モニタリングなどのビジネスを構築し、リアルタイムデータウェアハウスソリューションを形成しています（図 1 を参照）。これらのサービスはデータの正確性について厳格な要件があるため、リアルタイムデータウェアハウスの全パイプラインでエンドツーエンドの Exactly-Once を確保する必要があります。

一般的に、Flink の上流はプルベースの永続ストレージ（Kafka など）であり、繰り返し読み取りまたは消費できます。ソース側で Exactly-Once を実現するには、ソース側の読み取り位置を巻き戻すだけで済みます。シンク側の Exactly-Once はより複雑です。シンクはプッシュベースであり、出力先システムのトランザクション保証に依存する必要がありますが、コミュニティ版 ClickHouse はトランザクションをサポートしていません。

この状況に対応するため、Alibaba Cloud EMR ClickHouse と Flink チームは深く共同開発に取り組み、Flink から ClickHouse への Exactly-Once 書き込みをサポートし、リアルタイムデータウェアハウス全体の正確性を確保しました。本記事では、既存のメカニズムと実装方式をそれぞれ紹介します。

2、メカニズムの整理

ClickHouse の書き込みメカニズム

ClickHouse は MPP アーキテクチャの列指向 OLAP システムです（図 2 を参照）。各ノードはピア関係にあり、ZooKeeper の連携により、各ノードにローカルテーブルを同時書き込みすることで大規模なデータインポートが可能です。

ClickHouse のデータパートは、データストレージの最小単位です。ClickHouse が受信したデータブロックが書き込まれる際、パーティションの粒度で分割され、1 つ以上のデータパートが形成されます。データパートがディスクに書き込まれた後、バックグラウンドマージスレッドによって小規模なデータパートが大規模なデータパートにマージされ、ストレージと読み取りのコストが削減されます。

ローカルテーブルにデータを書き込む際、ClickHouse はまず一時的なデータパートを書き込みます。この時点ではデータはクライアントから参照できません。その後、rename 操作により一時データパートを正式なデータパートに昇格させ、データがクライアントから参照可能になります。ほとんどの一時データパートは迅速かつ正常に正式なデータパートへ rename されます。rename に失敗した一時データパートは、最終的に ClickHouse のクリーンアップポリシーによってディスクから削除されます。

以上の分析から、ClickHouse には一時データパートから正式なデータパートへのデータ書き込みメカニズムが存在することがわかります。このメカニズムを改修して二相コミットプロトコルに対応させることができます。二相コミットプロトコルは、分散システムでトランザクションコミットの整合性を実現するための重要なプロトコルです。

注：複数の Flink タスクが同じシャードまたはレプリカに書き込み可能です。

Flink の書き込みメカニズム

Flink は分散処理エンジンとして、トランザクションベースのシンクメカニズムを提供し、書き込みの Exactly-Once を保証できます。対応するデータ受信側は、XA 仕様に準拠した JDBC を提供する必要があります。完全な XA 仕様は非常に複雑なため、まず Flink の処理メカニズムを整理し、ClickHouse の実情に基づいて実装すべきインターフェイスの範囲を確定します。

分散書き込みでの統一されたトランザクションコミットを実現するため、Flink はチェックポイントメカニズムを利用します。このメカニズムは、各オペレーターの状態のスナップショットを定期的に生成し、永続化できます。チェックポイントメカニズムでは、コーディネーターがすべてのオペレーターの動作を調整する役割を担います。オペレーターの視点では、チェックポイントは初期化 → スナップショット生成 → チェックポイント完了/破棄の 3 つのフェーズがあります。コーディネーターの視点では、チェックポイントは定期的にトリガーする必要があり、すべてのオペレーターがスナップショットを完了した後に完了通知をトリガーする必要があります（付録 1 を参照）。

次に、Flink のオペレーターがトランザクションとチェックポイントメカニズムを利用して Exactly-Once を保証する方法を紹介します。オペレーターの完全な実行には、初期、データ書き込み、スナップショット、コミット、クローズの各ステージを経る必要があります。

初期ステージ：

・前回のタスク実行時のスナップショットから永続化された XID レコードを抽出します。スナップショットには主に 2 種類の XID が格納されています。1 つはスナップショットフェーズを完了していない XID、もう 1 つはスナップショットを完了した XID です。

・次に、前回スナップショットを完了しなかった XID に対してロールバック操作を実行します。前回スナップショットを完了したがコミットに失敗した XID に対しては、コミットのリトライ操作を実行します。

・上記の操作が失敗した場合、タスクの初期化に失敗し、タスクを中止してクローズフェーズに入ります。上記の操作が成功した場合は続行します。

・新しい一意の XID をトランザクション ID として生成し、スナップショットに記録します。

・新しく生成した XID を使用して、JDBC が提供する start() インターフェイスを呼び出します。

データ書き込みステージ：

・トランザクションが開始されると、データ書き込みのフェーズに入ります。オペレーターはほとんどの時間、このフェーズにいます。ClickHouse とのやり取りでは、このフェーズで JDBC が提供する preparedStatement の addBatch() および executeBatch() インターフェイスを呼び出します。データが書き込まれるたびに、現在の XID がメッセージに含まれます。

・データ書き込みステージでは、まずデータをオペレーターのメモリに書き込み、メモリ内のバッチデータを以下の 3 つの方法で ClickHouse に送信します。メモリ内のデータ数が batchSize のしきい値に達した場合、バックグラウンドタイマースレッドが一定間隔で自動フラッシュをトリガーした場合、スナップショットフェーズで end() および prepare() インターフェイスを呼び出す前にフラッシュを呼び出してキャッシュをクリアした場合です。

スナップショットフェーズ：

・現在のトランザクションは end() および prepare() インターフェイスを呼び出し、コミットを待機してスナップショット内のステータスを更新します。

・次に、このタスクの次の XID として新しいトランザクションを開き、新しいトランザクションをスナップショットに記録し、JDBC が提供する start() インターフェイスを呼び出して新しいトランザクションを開始します。

・スナップショットをストレージに永続化します。

完了ステージ：

すべてのオペレーターのスナップショットフェーズが正常に完了した後、コーディネーターはすべてのオペレーターにチェックポイントの成功を通知します。ClickHouse とのやり取りでは、このフェーズでオペレーターが JDBC が提供する commit() インターフェイスを呼び出してトランザクションをコミットします。

クローズフェーズ：

・現在のトランザクションがスナップショットフェーズに達していない場合、現在のトランザクションをロールバックします。

・すべてのリソースをクローズします。

以上のプロセスから、Flink はチェックポイントとトランザクションメカニズムを通じて、上流のデータをチェックポイント周期に従ってバッチに分割し、各バッチのデータが完全に書き込まれることを保証し、その後コーディネーターがすべてのオペレーターにコミット操作の完了を通知することがわかります。オペレーターの書き込みに失敗した場合、最後に成功したチェックポイントの状態に戻り、スナップショットに記録された XID に基づいて当該バッチのチェックポイントの XID をすべてロールバックします。コミット操作に失敗した場合、コミット操作がリトライされ、それでも失敗した場合は人為的な介入で対処します。

3、技術方案

全体方案

Flink と ClickHouse の書き込みメカニズムに基づき、Flink から ClickHouse へのトランザクション書き込みのシーケンス図を描くことができます（図 3 を参照）。ClickHouse のローカルテーブルへの書き込みであり、トランザクションの統一コミットはコーディネーターによって保証されるため、ClickHouse は XA 仕様の標準的な分散トランザクションを実装する必要がなく、二相コミットプロトコルのいくつかの重要なインターフェイスのみを実装すればよく、その他のインターフェイスは JDBC 側でデフォルト値を使用できます。

図 3 Flink から ClickHouse へのトランザクション書き込みのシーケンス図

ClickHouse-Server

状態機械

ClickHouse のトランザクションを実装するため、まずトランザクションが実装すべき許可された操作をいくつか定義します。

・Begin：トランザクションを開始します。

・Write Data：トランザクション内でデータを書き込みます。

・Commit：トランザクションをコミットします。

・Rollback：コミットされていないトランザクションをロールバックします。

トランザクションステータス：

・Unknown：トランザクションが開かれていません。この状態でいずれの操作も実行すると不正です。

・Initialized：トランザクションが開かれており、この時点ですべての操作が許可されています。

・Committing：トランザクションがコミット中です。Begin/Write Data 操作は許可されません。

・Committed：トランザクションがコミット済みで、操作は許可されません。

・Aborting：トランザクションがロールバック中で、操作は許可されません。

・Aborted：トランザクションがロールバック済みで、操作は許可されません。

図中のすべての操作は冪等です。その中で、Committing から Committed への遷移と、Aborting から Aborted への遷移では、いかなる操作も実行する必要はありません。Commit または Rollback の実行を開始すると、トランザクションステータスは Committing または Aborting に変化します。Commit または Rollback の実行後、トランザクションステータスは Committed または Aborted に設定されます。

Transaction

クライアントは HTTP RESTful API を通じて ClickHouse Server にアクセスします。クライアントと ClickHouse Server 間の完全なトランザクションのやり取りプロセスを図 5 に示します。

正常フロー：

・クライアントは、ClickHouse クラスター内の任意の ClickHouse Server に Begin Transaction リクエストを送信し、クライアントが生成したグローバル一意の Transaction ID を含めます。ClickHouse Server は Begin Transaction リクエストを受信すると、Transaction ID を ZooKeeper に登録し（Transaction ID と子 Znode の作成を含む）、トランザクションのステータスを Initialized に初期化します。

・クライアントが Begin Transaction の成功レスポンスを受信すると、データの書き込みを開始できます。ClickHouse Server はクライアントから送信されたデータを受信すると、一時データパートを生成しますが、正式なデータパートには変換しません。ClickHouse Server は、書き込まれた一時データパートの情報を JSON フォーマットで ZooKeeper 上のトランザクション情報に記録します。

・クライアントはデータの書き込みを完了した後、ClickHouse Server に Commit Transaction リクエストを送信します。ClickHouse Server は Commit Transaction リクエストを受信すると、ZooKeeper 上の該当トランザクションのデータパート情報に基づいて、ローカルの一時データパートを正式なデータパートに変換し、トランザクションステータスを Committed に更新します。Rollback のプロセスも Commit と同様です。

例外処理：

・Transaction ID の作成時に ZooKeeper で同じ Transaction ID が見つかった場合、ZooKeeper に記録されたトランザクションステータスに従って処理します。ステータスが Unknown の場合は処理を続行します。ステータスが Initialized の場合はそのまま返します。それ以外の場合は例外をスローします。

・現在実装されているトランザクションは分散トランザクションをサポートしておらず、スタンドアロントランザクションのみをサポートしています。したがって、クライアントは Transaction ID を登録した ClickHouse Server ノードにのみデータを書き込むことができます。当該ノード以外の ClickHouse Server がそのトランザクションのデータを受信した場合、エラーメッセージを返します。

・データの書き込みとは異なり、Commit フェーズでクライアントが Transaction ID を登録していない ClickHouse Server に Commit Transaction リクエストを送信した場合、ClickHouse Server はエラーメッセージを返さず、Transaction ID を登録している ClickHouse Server のアドレスをクライアントに返し、クライアントが正しい ClickHouse Server にリダイレクトできるようにします。Rollback のプロセスも Commit と同様です。

ClickHouse-JDBC

XA 仕様によると、完全な分散トランザクションメカニズムには多数の標準インターフェイスの実装が必要です（付録 2 を参照）。本設計では、いくつかの重要なインターフェイスのみを実装すればよいため、コンポジションベースのアダプタパターンを採用し、標準 XA インターフェイスに基づく XAResource 実装を Flink に提供します。同時に、ClickHouse Server がサポートする必要のないインターフェイスをシールドします。

XADataSource の実装には、継承ベースのアダプタパターンを採用し、Exactly-Once 機能に応じてデフォルト設定の一部を変更しています。たとえば、送信失敗時のリトライ回数などです。

さらに、本番環境では、通常 SLB（Server Load Balancer）を通じてデータ書き込みの負荷分散を行いますが、分散テーブルは使用しません。Exactly-Once のシナリオでは、Flink 側のタスクは特定の ClickHouse Server ノードへの接続を維持する必要があるため、Server Load Balancer を負荷分散に使用できません。この問題を解決するため、BalanceClickHouseDataSource のアプローチを参考にし、URL に複数の IP を設定し、properties で write_mode を Random に設定することで、Exactly-Once を確保しつつ XADataSource に負荷分散機能を持たせています。

Flink-Connector-ClickHouse

Flink はストリーミングデータ処理エンジンとして、複数のデータ受信先への書き込み機能をサポートしており、各受信先は特定の connector を実装する必要があります。Exactly-Once に対応するため、ClickHouse Connector は XADataSource のオプション設定を追加し、クライアント設定に基づいて Exactly-Once 機能を提供します。

4、テスト結果

ClickHouse トランザクション性能テスト

・ClickHouse の単一バッチへのデータ書き込み量と総バッチ数を同一にし、クライアント側の同時書き込みスレッド数を変えた場合の性能を図 6 に示します。ClickHouse のトランザクション有効/無効にかかわらず、ClickHouse のスループットはクライアント側の同時書き込みスレッド数に正比例します。トランザクションを有効にすると、ClickHouse 内の一時データパートは即座に正式なデータパートに変換されないため、トランザクション完了前に多数の一時データパートが ClickHouse のマージプロセスに参加せず、ディスク IO の書き込み性能への影響が軽減されます。そのため、トランザクション有効時の書き込み性能はトランザクション無効時よりも優れています。ただし、トランザクションに含まれるバッチ数が増加すると、ディスク上の一時データパートの増加により統合時の CPU 負荷が増大し、書き込み性能に影響が出て、トランザクション有効時の書き込み性能も低下します。

・ClickHouse への総書き込みバッチ数とクライアント側の同時書き込みスレッド数を同一にし、単一バッチのデータ量を変えた場合の性能を図 7 に示します。

ClickHouse のスループットは、トランザクションの有効/無効にかかわらず、単一バッチのデータ量に正比例します。トランザクションを有効にした場合、各バッチのデータ量が少ないほど、トランザクションの有無が ClickHouse のスループットに与える影響は大きくなります。これは、各バッチの書き込み時間がトランザクション処理の中で相対的に小さくなり、トランザクションの影響を受けやすくなるためです。したがって、トランザクションに含まれるバッチ数が多いほど、トランザクションが書き込み性能に与える影響は小さくなります。トランザクションに含まれるバッチ数が増えると、トランザクション処理時間の書き込みにおける割合が徐々に低下し、ClickHouse のマージ処理の影響が増大するため、書き込み性能に影響が出ます。トランザクション有効時の性能はトランザクション無効時よりも優れています。

・全体的に、トランザクションの有効化が書き込み性能に与える影響は限定的であり、期待に沿った結果です。

Flink から ClickHouse への書き込み性能比較

・同一のデータ量で異なるチェックポイント周期の場合、Flink から ClickHouse への書き込みに要した総時間を図 8 に示します。チェックポイント周期は、Exactly-Once を有効にしていないタスクの処理時間には影響しないことがわかります。Exactly-Once を有効にしているタスクでは、5 秒から 60 秒の範囲で処理時間がまず減少し、その後増加する傾向を示します。理由は、チェックポイント周期が短い場合、Exactly-Once を有効にしたオペレーターと ClickHouse 間のトランザクション相互作用が頻繁になるためです。チェックポイント周期が長い場合、Exactly-Once を有効にしたオペレーターは最後のトランザクションをコミットしてデータを可視化するためにチェックポイント周期の終了を待つ必要があるためです。本テストのチェックポイント周期データは参考値としてご利用ください。本番環境では、マシン仕様とデータ書き込み速度に応じて調整する必要があります。

・全体的に、Flink が ClickHouse に書き込む際に Exactly-Once 機能を有効にすると、性能に若干の影響があります。この結論は期待に沿ったものです。

5、今後の計画

本バージョンの EMR ClickHouse で実装されたトランザクションは、まだ完全とは言えません。スタンドアロントランザクションのみをサポートしており、分散トランザクションはサポートしていません。分散システムでは一般的に、Meta Server を使用してメタデータを一元的に管理し、分散トランザクションメカニズムをサポートします。現在、ClickHouse MetaServer を設計して分散トランザクションをサポートし、ClickHouse の ZooKeeper への依存を排除できるようにする計画を進めています。