メインコンテンツへスキップ

​
はじめに

ClickHouse はクエリを非常に高速に処理しますが、これらのクエリは 複数のサーバー間でどのように分散され、並列化されているのでしょうか。
このガイドでは、まず ClickHouse が分散テーブルを介してクエリを 複数の分片にどのように分散するのかを説明し、その後、クエリの実行で 複数のレプリカをどのように活用できるのかを説明します。

​
分片アーキテクチャ

shared-nothing アーキテクチャでは、クラスターは通常、複数の分片に 分割され、各分片には全体データの一部が格納されます。これらの分片の 上位には分散テーブルがあり、データ全体を単一のビューとして提供します。 読み取りはローカルテーブルに送ることができます。その場合、クエリの実行は 指定した分片でのみ行われます。あるいは分散テーブルに送ることもでき、その 場合は各分片で指定されたクエリが実行されます。分散テーブルに対するクエリを 受けたサーバーがデータを集約し、クライアントに応答を返します。 上の図は、クライアントが分散テーブルにクエリを実行した際に何が起こるかを示しています。
  1. SELECT クエリは、任意のノード上の分散テーブルに送信されます (ラウンドロビン戦略を介するか、ロードバランサーによって特定のサーバーに ルーティングされた後) 。このノードがコーディネーターとして動作します。
  2. このノードは、分散テーブルで指定された情報に基づいて、 クエリを実行する必要がある各分片を特定し、クエリを 各分片に送信します。
  3. 各分片はローカルでデータを読み取り、フィルタリングと集約を行った後、 マージ可能な状態をコーディネーターに返します。
  4. コーディネーターとなるノードがデータをマージし、その後 応答をクライアントに返します。
ここにレプリカが加わっても、処理の流れはほぼ同じで、違いは 各分片から 1 つのレプリカだけがクエリを実行することです。 つまり、より多くのクエリを並列に処理できるようになります。

​
非分片アーキテクチャ

ClickHouse Cloud のアーキテクチャは、上で示したものとは大きく異なります。 (詳細は “ClickHouse Cloud Architecture” を参照してください) 。コンピュートとストレージが分離され、さらに事実上 無制限のストレージを利用できるため、分片の必要性は以前ほど高くありません。 以下の図は ClickHouse Cloud のアーキテクチャを示しています。 このアーキテクチャでは、レプリカをほぼ瞬時に追加・削除できるため、 クラスターの高い拡張性を実現できます。ClickHouse Keeper クラスター (右側に表示) は、メタデータの単一の信頼できる情報源を 担います。レプリカは ClickHouse Keeper クラスターからメタデータを取得し、すべて同じデータを維持できます。データ自体は オブジェクトストレージに保存され、SSD cache によってクエリを高速化できます。 では、クエリ実行を複数のサーバーにまたがってどのように分散するのでしょうか。分片 アーキテクチャでは、各分片がデータの一部に対して実際にクエリを実行できるため、 比較的わかりやすいものでした。では、sharding がない場合はどのように機能するのでしょうか。

​
並列レプリカの紹介

複数のサーバーにまたがってクエリ実行を並列化するには、まず サーバーのうち 1 台をコーディネーターとして割り当てられるようにする 必要があります。コーディネーターは、実行すべき タスクの一覧を作成し、それらがすべて実行・ 集計され、結果がクライアントに返されることを保証する役割を担います。多くの 分散システムと同様に、通常これは最初の クエリを受け取ったノードが担います。また、作業単位を定義する必要もあります。分片化アーキテクチャでは、 作業単位は分片、つまりデータの一部分です。並列レプリカでは、 作業単位として グラニュール と呼ばれるテーブルの小さな部分を 使用します。 では、以下の図を使って、実際にどのように動作するのか見てみましょう。 並列レプリカでは:
  1. クライアントからのクエリは、ロード バランサーを経由して 1 つのノードに送られます。このノードがこのクエリのコーディネーターになります。
  2. そのノードは各パーツの索引を解析し、処理対象のパーツと グラニュールを選択します。
  3. コーディネーターは、ワークロードを 異なるレプリカに割り当て可能な一連のグラニュールに分割します。
  4. 各グラニュールの集合は対応するレプリカで処理され、完了すると マージ可能な状態がコーディネーターに送られます。
  5. 最後に、コーディネーターがレプリカからのすべての結果をマージし、 クライアントにレスポンスを返します。
上記の手順は、並列レプリカが理論上どのように動作するかを示しています。 しかし実際には、このような仕組みが 完全には機能しない要因が数多くあります。
  1. 利用できないレプリカがある場合があります。
  2. ClickHouse のレプリケーションは非同期であるため、ある時点では一部のレプリカが 同じパーツを持っていない可能性があります。
  3. レプリカ間のテールレイテンシを何らかの方法で扱う必要があります。
  4. ファイルシステムキャッシュは各レプリカ上の アクティビティに応じて異なるため、タスクをランダムに割り当てると、 キャッシュ局所性の観点では最適でないパフォーマンスにつながる可能性があります。
これらの要因をどのように克服しているかは、以降のセクションで説明します。

​
アナウンス

上のリストの (1) と (2) に対処するため、アナウンス という概念を導入しました。以下の図でその仕組みを見てみましょう。
  1. クライアントからのクエリは、ロード バランサーを経由して 1 つのノードに送られます。このノードが、このクエリのコーディネーターになります。
  2. コーディネーターとなるノードは、 クラスター内のすべてのレプリカからアナウンスを取得するためのリクエストを送信します。レプリカごとに、テーブルの現在のパーツ集合の見え方が わずかに異なる場合があります。そのため、 不正確なスケジューリング判断を避けるには、この情報を収集する必要があります。
  3. その後、コーディネーターとなるノードはアナウンスを使って、 各レプリカに割り当て可能な グラニュール の集合を定義します。たとえばここでは、 パーツ 3 の グラニュール はレプリカ 2 には 1 つも割り当てられていないことがわかります。 これは、このレプリカがアナウンスでこのパーツを通知しなかったためです。 また、 レプリカ 3 はアナウンスを返さなかったため、タスクが 1 つも割り当てられていない点にも注意してください。
  4. 各レプリカが自分に割り当てられた グラニュール のサブセットに対してクエリを処理し、 マージ可能な状態がコーディネーターに送り返されると、 コーディネーターが結果をマージし、そのレスポンスがクライアントに送信されます。

​
動的協調

テールレイテンシの問題に対処するため、動的協調を追加しました。これは、 すべてのグラニュールを1回のリクエストでレプリカに送るのではなく、各レプリカが コーディネーターに新しいタスク (処理するグラニュールのセット) を要求 できることを意味します。コーディネーターは、受け取ったアナウンスに基づいて グラニュールのセットをレプリカに割り当てます。 ここでは、すべてのレプリカがすべてのパーツを含む アナウンスを送信し終えた段階にあると仮定します。 以下の図は、動的協調がどのように機能するかを示しています。
  1. レプリカは、タスクを処理できることをコーディネーターノードに知らせます。 また、どれだけの作業を処理できるかを指定することもできます。
  2. コーディネーターはレプリカにタスクを割り当てます。
  1. レプリカ1と2は、自分たちのタスクを非常にすばやく完了します。 そのため、コーディネーターノードに別のタスクを要求します。
  2. コーディネーターはレプリカ1と2に新しいタスクを割り当てます。
  1. すべてのレプリカが、それぞれのタスクの処理を完了しました。 さらにタスクを要求します。
  2. コーディネーターは、アナウンスを使って未処理のタスクが残っているかどうかを 確認しますが、残っているタスクはありません。
  3. コーディネーターは、すべての処理が完了したことをレプリカに伝えます。 その後、マージ可能なすべての状態をマージして、クエリに応答します。

​
cache の局所性の管理

最後に残っている潜在的な問題は、cache の局所性をどう扱うかです。クエリが複数回実行される場合、同じタスクが同じレプリカにルーティングされることを、どのように保証すればよいのでしょうか。前の例では、次のタスクが割り当てられていました。
レプリカ 1レプリカ 2レプリカ 3
パート 1g1, g6, g7g2, g4, g5g3
パート 2g1g2, g4, g5g3
パート 3g1, g6g2, g4, g5g3
同じタスクが同じレプリカに割り当てられ、cache の恩恵を受けられるようにするため、2 つの処理が行われます。まず、パーツ + グラニュール の集合 (タスク) のハッシュを計算します。次に、タスクの割り当てに対して、レプリカ数を法とする剰余演算を適用します。 理屈の上ではよさそうに見えますが、実際には、あるレプリカに急に負荷がかかったり、ネットワークが劣化したりすると、特定のタスクの実行に同じレプリカが継続的に使われることで、テールレイテンシが発生する可能性があります。max_parallel_replicas がレプリカ数より少ない場合は、クエリ実行のためにランダムにレプリカが選択されます。

​
タスクのスティーリング

あるレプリカでのタスク処理が他のレプリカより遅い場合、他のレプリカは、ハッシュに基づいて本来そのレプリカに割り当てられているタスクを「奪う」ことで、テールレイテンシの低減を図ります。

​
制限事項

この機能には既知の制限があり、主なものをこのセクションで説明します。
以下に挙げた制限事項以外の問題が見つかり、その原因が並列レプリカにあると思われる場合は、 GitHub でラベル comp-parallel-replicas を付けて issue を作成してください。

​
並列レプリカの問題の調査

各クエリでどの設定が使われているかは、 system.query_log テーブルで確認できます。また、 system.events テーブルを見ると、サーバー上で発生したすべてのイベントを確認できます。さらに、 clusterAllReplicas テーブル関数を使うと、すべてのレプリカ上のテーブルを確認できます (Cloud ユーザーの場合は default を使用してください) 。
Query
Response
system.text_log テーブルには、並列レプリカを使用したクエリの実行に関する情報も 含まれています。
Query
Response
最後に、EXPLAIN PIPELINE を使用することもできます。これにより、ClickHouse がクエリをどのように実行し、クエリの実行に どのようなリソースが使われるかを確認できます。例として、次のクエリを見てみましょう。
並列レプリカを使わない場合のクエリパイプラインを見てみましょう。
EXPLAIN PIPELINE (without parallel replica)
続いて、並列レプリカ を有効にした場合:
EXPLAIN PIPELINE (with parallel replica)
最終更新日 2026年7月3日