ByteHouseリアルタイムインポート技術の進化

ByteHouse は Volcano Engine 上のクラウドネイティブデータウェアハウスであり、ユーザーに非常に高速な分析エクスペリエンスを提供し、リアルタイムデータ分析と大規模なオフラインデータ分析をサポートします。便利な弾力的なスケーリング機能、卓越した分析パフォーマンス、豊富なエンタープライズレベルの機能を備えており、顧客のデジタル変革を支援します。

この記事では、需要動機、技術的実装、実用化の観点から、さまざまなアーキテクチャに基づく ByteHouse リアルタイムインポートテクノロジーの進化を紹介します。

社内業務のリアルタイムインポート要件

ByteHouse のリアルタイムインポートテクノロジーの進化のきっかけは、当初 ByteDance の社内ビジネスのニーズでした。

ByteHouse 内では、Kafka は依然としてリアルタイムインポートの主なデータソースです (この記事では Kafka インポートを例として使用しており、以下では繰り返しません)。ほとんどの社内ユーザーにとって、データの量は比較的多くなります。したがって、ユーザーは、データのインポートのパフォーマンス、サービスの安定性、およびインポート機能のスケーラビリティにさらに注意を払うことになります。データの遅延に関しては、データが数秒以内に表示されれば、ほとんどのユーザーのニーズは満たされます。このようなシナリオに基づいて、ByteHouse はカスタマイズされた最適化を実行します。

分散アーキテクチャによる高可用性

コミュニティネイティブ分散アーキテクチャ

ByteHouse は最初に ClickHouse コミュニティの分散アーキテクチャを採用しましたが、分散アーキテクチャには固有のアーキテクチャ上の欠陥がいくつかあります。これらの問題点は主に次の 3 つの側面で現れます。

ノード障害: クラスターマシンの数が一定の規模に達すると、ほぼ毎週、ノード障害を手動で処理する必要があります。単一コピークラスターの場合、極端なケースでは、ノード障害によってデータが失われる可能性もあります。
読み取り/書き込み競合: 分散アーキテクチャの読み取り/書き込み結合により、クラスターの負荷が特定のレベルに達すると、ユーザークエリとリアルタイムインポートでリソースの競合 (特に CPU と IO) が発生します。輸入に影響が出て消費の遅れが生じます。
拡張コスト: 分散アーキテクチャのデータは基本的にローカルに保存されるため、拡張後にデータを入れ替えることはできません。新しく拡張されたマシンにはほとんどデータが残っていない一方で、古いマシンのディスクはほぼいっぱいになっている可能性があり、その結果、クラスターの負荷が不均一になり、拡張が非効率的になります。

これらは分散アーキテクチャの自然な問題点ですが、その自然な同時実行特性と、ローカルディスクデータの読み取りと書き込みの究極のパフォーマンス最適化により、利点と欠点の両方があると言えます。

コミュニティによるデザインのリアルタイムインポート

高レベル消費モード: 消費負荷のバランスをとるために、Kafka 独自の再バランスメカニズムに依存します。
2レベルの同時実行

分散アーキテクチャに基づくリアルタイムインポートのコア設計は、実際には 2 レベルの同時実行性です。

CH クラスターには通常複数のシャードがあり、各シャードは消費インポートを同時に実行します。これは、第 1 レベルのシャード間のマルチプロセスの同時実行です。

各シャード内で複数のスレッドを同時に使用することもできるため、非常に高いパフォーマンスのスループットを実現できます。

バッチ書き込み

単一のスレッドの場合、基本的な消費パターンはバッチ書き込みです。つまり、一定量のデータを消費するか、一定期間後にすべてを一度に書き込みます。バッチ書き込みにより、パフォーマンスの最適化がより適切に実現され、クエリパフォーマンスが向上し、バックグラウンドのマージスレッドへの負荷が軽減されます。

満たされていないニーズ

上記のコミュニティの設計と実装では、ユーザーの高度なニーズをまだ満たすことができません。

まず、一部の上級ユーザーは、データ配布に関して比較的厳しい要件を持っています。たとえば、特定のデータには特定のキーがあり、同じキーを持つデータが同じシャードに保存されることが期待されます (一意のキー要件など)。この場合、コミュニティのハイレベルの消費パターンを満たすことはできません。
第二に、高レベルの消費フォームの再バランスは制御不能であり、最終的には ClickHouse クラスター内のインポートされたデータがシャード間で不均等に分散される可能性があります。
もちろん、消費タスクの分布は予測不可能であり、異常な消費シナリオでは、問題のトラブルシューティングが非常に困難になります。エンタープライズレベルのアプリケーションでは、これは受け入れられません。

自社開発の分散アーキテクチャ消費エンジンHaKafka

上記の要件を満たすために、ByteHouse チームは分散アーキテクチャに基づく消費エンジンである HaKafka を開発しました。

高可用性（Ha）

HaKafka は、コミュニティのオリジナルの Kafka テーブルエンジンの消費上の利点を継承し、高可用性の Ha 最適化に重点を置いています。

分散アーキテクチャの観点から見ると、実際には各シャードに複数のレプリカが存在する可能性があり、各レプリカに HaKafka テーブルを作成できます。ただし、ByteHouse は ZK を通じてリーダーのみを選択し、リーダーが実際に消費プロセスを実行できるようにし、他のノードはスタンバイ状態になります。リーダーノードが利用できない場合、ZK は数秒以内にリーダーをスタンバイノードに切り替えて消費を継続できるため、高い可用性が実現します。

低レベル消費モデル

HaKafka の消費モードは、高レベルモードから低レベルモードに調整されます。低レベルモードでは、トピックパーティションがクラスター内の各シャードに整然と均等に分散されることを保証できます。同時に、シャード内で複数のスレッドを再利用できるため、各スレッドが異なるパーティションを消費できるようになります。これは、コミュニティ Kafka テーブルエンジンの 2 レベルの同時実行性の利点を完全に継承します。

低レベル消費モードでは、上流ユーザーがトピックへの書き込み時にデータの偏りがないことを確認する限り、HaKafka を介して Clickhouse にインポートされたデータは確実にシャード全体に均等に分散されます。

同時に、同じキーを持つデータを同じシャードに書き込むなど、特別なデータ分散要件を持つ上級ユーザーの場合、アップストリームが同じキーを持つデータが同じパーティションに書き込まれることを保証している限り、ByteHouse をインポートすることでユーザーのニーズを完全に満たし、一意のキーなどのシナリオを適切にサポートできます。

シナリオ 1:

上記の図に基づくと、デュアルレプリカシャードがある場合、各レプリカには準備完了状態の同一の HaKafka テーブルが含まれます。ただし、HaKafka は、ZK によって正常に選出されたリーダーノード上でのみ、対応する消費プロセスを実行します。リーダーノードがダウンすると、レプリカ 2 が新しいリーダーとして自動的に選択され、消費を継続するため、高可用性が確保されます。

シナリオ2:

ノード障害が発生した場合、通常はノード交換プロセスを実行する必要があります。分散ノードの置き換えには、データのコピーという非常に重い操作が伴います。

マルチレプリカクラスターの場合、1 つのレプリカに障害が発生しても、他のレプリカはそのまま残ります。当然ながら、ノードの置き換えフェーズでは、古いデータが完全であるため、Kafka の消費が完全なレプリカ Replica 2 に配置されることを期待します。このように、レプリカ 2 は常に完全なデータセットとなり、外部サービスを正常に提供できるようになります。 HaKafka はこれを保証できます。 HaKafka がリーダーを選出する際、特定のノードがノードを置き換えている途中であると判断された場合、そのノードはリーダーとして選択されません。

インポートパフォーマンスの最適化: メモリテーブル

HaKafka はメモリテーブルも最適化します。

企業が数百の列または数千のマップキーを持つ大規模なテーブルを持っているシナリオを考えてみましょう。 ClickHouse の各列は特定のファイルとしてディスクに書き込まれるため、列の数が増えると、インポートするたびに書き込まれるファイルも多くなります。すると、同じ消費時間内に、断片化されたファイルが多数頻繁に書き込まれることになり、マシンの IO に大きな負担がかかり、MERGE に大きな圧力がかかります。深刻な場合には、クラスターが使用できなくなる可能性もあります。このシナリオを解決するために、インポートパフォーマンスを最適化するメモリテーブルを設計しました。

メモリテーブルアプローチでは、データがインポートされるたびにディスクに直接フラッシュされるのではなく、メモリにデータを保存します。データ量が一定量に達すると、IO 操作を削減するためにバッチでディスクにフラッシュされます。メモリテーブルは外部クエリサービスを提供できます。クエリは、コンシューマーノードが配置されているレプリカにルーティングされ、メモリテーブル内のデータが読み取られるため、データのインポートの遅延は影響を受けません。社内の使用経験に基づくと、Memory Table は、一部の大規模で幅の広いテーブルのビジネスインポートニーズを効果的に解決するだけでなく、インポートパフォーマンスを最大約 3 倍向上させます。

新しいクラウドネイティブアーキテクチャ

上で説明した分散アーキテクチャの固有の欠陥を考慮して、ByteHouse チームはアーキテクチャのアップグレードに取り組んできました。当社では、ビジネスの主流であるクラウドネイティブアーキテクチャを選択しました。新しいアーキテクチャは2021年初頭にByteDanceの社内業務に使用され始め、コードは2023年初頭にオープンソース化されました[ByConity] https://github.com/ByConity/ByConity。

クラウドネイティブアーキテクチャ自体には、自然な自動フォールトトレランスと軽量な拡張および縮小機能が備わっています。同時に、データがクラウドに保存されるため、ストレージとコンピューティングの分離が実現され、データのセキュリティと安定性も向上します。もちろん、クラウドネイティブアーキテクチャには欠点がないわけではありません。元のローカル読み取りと書き込みをリモート読み取りと書き込みに変更すると、必然的に読み取りと書き込みのパフォーマンスに一定の損失が生じます。しかし、ある程度のパフォーマンス損失と引き換えにアーキテクチャの合理性を確保し、運用・保守コストを削減することは、実際にはデメリットよりもメリットの方が多いのです。

上図は、ByteHouse クラウドネイティブアーキテクチャのアーキテクチャ図です。この記事では、リアルタイムインポートに関連する重要なコンポーネントをいくつか紹介します。

クラウドサービス

まず、全体のアーキテクチャは 3 つのレイヤーに分かれています。最初のレイヤーはクラウドサービスであり、主にサーバーと Catlog の 2 つのコンポーネントで構成されます。このレイヤーはサービスの入り口であり、クエリのインポートを含むすべてのユーザー要求はサーバーから入ります。サーバーはリクエストを前処理するだけで、具体的には実行しません。 Catlog でメタ情報を照会した後、前処理されたリクエストとメタ情報が実行のために仮想ウェアハウスに送信されます。

仮想倉庫

仮想ウェアハウスは実行レイヤーです。異なるビジネスは独立した仮想倉庫を持ち、リソースの分離を実現できます。現在、仮想ウェアハウスは主に「デフォルト」と「書き込み」の 2 つのカテゴリに分かれています。 Default は主にクエリに使用され、Write はインポートに使用され、読み取りと書き込みの分離を実現します。

仮想通貨

最下層は VFS (データストレージ) で、HDFS、S3、AWS などのクラウドストレージコンポーネントをサポートします。

クラウドネイティブアーキテクチャに基づくリアルタイムインポート設計

クラウドネイティブアーキテクチャでは、サーバーは特定のインポート実行を実行せず、タスク管理のみを実行します。したがって、サーバー側では、各コンシューマーテーブルにマネージャーがいて、すべてのコンシューマー実行タスクを管理し、仮想ウェアハウスで実行されるようにスケジュールします。

HaKafka の低レベル消費モードを継承しているため、マネージャーは設定された消費タスクの数に応じてトピックパーティションを各タスクに均等に分散します。消費タスクの数は構成可能であり、上限はトピックパーティションの数です。

上の図を見ると、マネージャーが左側にいることがわかります。カタログから対応するオフセットを取得し、指定された消費タスクの数に応じて対応する消費パーティションを割り当て、仮想ウェアハウスの異なるノードに実行するようにスケジュールします。

新しい消費実行プロセス

新しいクラウドネイティブアーキテクチャはトランザクションによって保証されるため、すべての操作が 1 つのトランザクション内で完了することが期待され、より合理的になります。

新しいクラウドネイティブアーキテクチャでのトランザクション実装に依存して、各消費タスクの消費プロセスには主に次の手順が含まれます。

消費が始まる前に、ワーカー側のタスクはまず、RPC リクエストを通じてサーバー側にトランザクションを作成するように要求します。
rdkafka::poll() を実行し、一定の時間（デフォルトでは8秒）または十分な大きさのブロックを消費する
ブロックをパーツに変換し、VFS にダンプします (この時点ではデータは表示されません)
トランザクションを開始する RPCリクエストを介してサーバーにリクエストをコミットする

(トランザクションでコミットされたデータには、完了したダンプ部分のメタデータと対応する Kafka オフセットが含まれます)

トランザクションの送信が成功しました（データが表示されます）

フォールトトレランス保証

上記の消費プロセスから、新しいクラウドネイティブアーキテクチャにおける消費のフォールトトレランスの保証は、主にマネージャーとタスク間の双方向ハートビートと高速障害戦略に基づいていることがわかります。

マネージャー自体は定期的なアクティビティ検出を実行し、スケジュールされたタスクが RPC を通じて正常に実行されているかどうかを確認します。
同時に、各タスクは消費中にトランザクション RPC 要求を利用して自身の有効性を検証します。検証が失敗した場合は自動的に強制終了されます。
マネージャーが活性状態の検出に失敗すると、第 2 レベルのフォールトトレランスを実現するために、直ちに新しい消費タスクが開始されます。

購買力

消費容量に関しては、前述の通り拡張可能です。消費タスクの数は、トピックパーティションの数までユーザーが設定できます。仮想ウェアハウス内のノード負荷が高い場合は、ノードを非常に簡単に拡張できます。

もちろん、マネージャーのスケジュールタスクは、リソースマネージャーを使用してタスクを管理およびスケジュールするという基本的な負荷分散保証を実装します。

セマンティック拡張: 正確に1回

最後に、新しいクラウドネイティブアーキテクチャにおける消費セマンティクスも、分散アーキテクチャの「少なくとも 1 回」から「正確に 1 回」に強化されました。

分散アーキテクチャにはトランザクションがないため、少なくとも 1 回しか実現できません。つまり、どのような状況でもデータが失われないことが保証されますが、極端な場合には重複した消費が発生する可能性があります。クラウドネイティブアーキテクチャでは、トランザクションの実装により、各消費でトランザクションを通じて Part と Offset のアトミック送信を実現できるため、Exactly-Once のセマンティック拡張が実現します。

メモリバッファ

HaKafka のメモリテーブルに対応して、クラウドネイティブアーキテクチャではメモリキャッシュメモリバッファーのインポートも実装されています。

メモリテーブルとは異なり、メモリバッファーは Kafka のコンシューマータスクにバインドされなくなり、ストレージテーブルのキャッシュレイヤーとして実装されます。これにより、Memory Buffer の汎用性が向上し、Kafka のインポートだけでなく、Flink のような小さなバッチインポートにも使用できるようになります。

同時に、新しいコンポーネント WAL を導入しました。データをインポートするときは、まず WAL を書き込みます。書き込みが成功すれば、データのインポートは成功したとみなされます。サービスが開始されると、ディスクにフラッシュされていないデータは最初に WAL から復元できます。次にメモリバッファに書き込みます。メモリバッファーはユーザーが照会できるため、正常に書き込まれたデータが表示されます。メモリバッファー内のデータも定期的にディスクにフラッシュされ、フラッシュ後に WAL からクリアできます。

ビジネスアプリケーションと将来の考え方

最後に、この記事では、ByteDance 内のリアルタイムインポートの現状と、次世代のリアルタイムインポートテクノロジーの最適化の方向性について簡単に紹介します。

ByteHouse のリアルタイムインポートテクノロジーは Kafka に基づいています。毎日のデータスループットはPBレベルです。単一のインポートスレッドまたは単一のコンシューマーのスループットの実験値は 10 ～ 20 MiB/s です。 (ここで強調する理由は、固定値やピーク値ではなく、経験値であるためです。消費スループットは、ユーザーテーブルの複雑さに大きく依存します。テーブル列の数が増えると、インポートパフォーマンスが大幅に低下する可能性があり、正確な計算式を使用することはできません。したがって、ここでの経験値は、Byte のほとんどのテーブルのインポートパフォーマンスの経験値に近いものです。)

ByteDance は、Kafka に加えて、RocketMQ、Pulsar、MySQL (MaterializedMySQL)、Flink 直接書き込みなど、他のデータソースのリアルタイムインポートもサポートしています。

次世代のリアルタイムインポートテクノロジーに関する簡単な考察: