*注: 同じクエリをループで継続的に再実行しています。初回の実行はあまりに速く終わるため目にとまらないほどです。速度を視覚的に確認できるよう、アニメーションではクエリを繰り返し実行しています。*
* **実行時間:** 0.414 秒
* **処理行数:** 1,000 億行
* **スループット:** 毎秒 2,418 億行 (1.45 TB/s)
## クエリの無限の水平スケーリング
このデモは小細工ではありません。私たちが長年追い求めてきた**クエリの無限の水平スケーリング**を解き放つ、ClickHouse Cloud の新機能 **[parallel replicas](https://clickhouse.com/docs/ja/products/cloud/features/infrastructure/parallel-replicas)** によって実現されています。
クラウドでのクエリのスケーリングは、ダイヤルを回すかのような感覚で行えます。**ノードを追加すれば、より高速な結果が得られます。**
これは実質的に、データの移動を一切伴わない*オンデマンドの仮想シャーディング*です。
現在 Cloud でベータ版 (まもなく一般提供 (GA)) として提供されているこの OSS 機能により、ClickHouse は 90 コアの単一マシンも、9,000 コアを備えた 100 台のマシンもまったく同様に扱えるようになります。
> 単一のクエリが 1 台のコンピュートノードに縛られることはなくなり、クラスター内の**全ノードの全コア**にまたがって実行できるようになります。
これは次のことを意味します。
* **弾力性** — ノードを追加すると、クエリが即座に高速化します。
* **シンプルさ** — データのリシャッフルは不要で、設定を有効にするだけです。
* **大規模環境でのインタラクティブな速度** — 1,000 億行以上を 1 秒未満で処理します。
Spark や、BigQuery、Snowflake といったクラシックなクラウドデータウェアハウスをご利用の方には、大規模並列処理 ([MPP](https://en.wikipedia.org/wiki/Massively_parallel)) を**より軽量にしたもの**として馴染み深く感じられるはずです。
現時点では、parallel replicas を GROUP BY のパフォーマンスにおける新たなベンチマークとして捉えてみてください。
分析速度を決定づけるこの処理は、割り当てられたコア数に応じて線形にスケールするようになりました。
## なぜ GROUP BY が分析の心臓部なのか
極めて洗練されたシステムの中には、1 つの役割を例外的なまでに完璧にこなすものがあります。万年筆は紙に優雅にインクを落とすために存在します。grep のような Unix ツールは 1 つの仕事を完璧に遂行します。ClickHouse もまったく同じ理念から始まりました。**何よりも高速に GROUP BY を実行すること**です。
> 「可能な限り高速にデータをフィルタリングして集計するという、たった 1 つの課題を解決することを目的としていました。言い換えれば、単に GROUP BY を実行するためです。」— ClickHouse の生みの親、Alexey Milovidov、[BDTC 2019 にて](https://www.youtube.com/watch?v=ZOZQCQEtrz8&t=103s)
数百万行や数十億行に対する集計を実行したとき、何かエラーが起きたのではないかと思うほど一瞬で完了する様子を想像してみてください。それは[私が初めて ClickHouse に触れたときの衝撃](https://www.youtube.com/watch?v=xqdoVSZYtT8)でもありました。「何かおかしいのでは？ こんなに速いわけがない。本当に？」
**GROUP BY はほぼあらゆる分析クエリの中核に位置しており**、[オブザーバビリティのダッシュボード](https://clickhouse.com/use-cases/observability)を支え、自然言語を SQL に変換する[会話速度の AI](https://www.linkedin.com/posts/alasdair-brown_how-does-your-choice-of-database-affect-your-activity-7359584224250585089-QgQ1) を加速し、[エージェント向け分析](https://clickhouse.com/ja/blog/agent-facing-analytics-jp)など、多岐にわたる用途を支えています。GROUP BY を高速化できれば、ほぼすべての分析ワークロードが高速化します。
> 実際の BI ワークロードを対象とした最近の VLDB 2025 の調査[によると](https://www.vldb.org/pvldb/vol18/p5126-bress.pdf)、**クエリの半数以上に GROUP BY が含まれており**、分析におけるその中心的な役割が浮き彫りになっています。
ClickHouse はそのために作られました。しかし、ワークロードが増大しクエリが複雑になるにつれて、期待値も高まりました。データ量が爆発的に増加し、クラスターが数百ノードに及ぶようになっても、エンジンは GROUP BY の高速性を維持し続けなければなりませんでした。
本記事では、この課題にどのように取り組んでいるかを解説します。まず手短なデモを行い、続いて GROUP BY を単一コアから数千コアへとクリーンにスケールさせる実行モデルを詳しく見ていきます。
## ClickHouse のスピードで GROUP BY を実行する
パフォーマンスの話は、ミリ秒やグラフの数値の中に埋もれてしまいがちです。代わりに、高速であるとはどういうことなのかを「体感」してみましょう。単一ノード上の数百万行からフリート全体に分散された数千億行まで、桁違いに大きくなっていくデータセットに対して同じ GROUP BY を実行します。
### まずは小さく始める: 数百万行の重要性
4 年前に ClickHouse で初めて集計を実行したとき、目にも留まらぬ速さで数百万行が処理されたため、何か障害でも起きたのではないかと思いました。
実際のワークロードにおいて数百万行は今でも重要です。そこで、まずは[英国の不動産価格データセット](https://clickhouse.com/docs/ja/get-started/sample-datasets/uk-price-paid) (英国政府によって継続的に更新されており、2025年8月時点で約 3,000 万行) を使ったシンプルな集計から始めましょう。このクエリは町ごとの販売額を合計し、上位 3 件を返します。
USE uk_base; -- ~30M rows
SELECT
town,
formatReadableQuantity(sum(price)) AS total_revenue
FROM uk_price_paid
GROUP BY town
ORDER BY sum(price) DESC
LIMIT 3
SETTINGS
enable_parallel_replicas = false;
### 単一ノード (89 コア) での 3,000 万行
ClickHouse Cloud に対し、parallel replicas を無効化した状態 (`enable_parallel_replicas = 0`) で clickhouse-client 経由で実行しました。
**注:** これは**同じクエリをループで継続的に再実行したもの**です。初回の実行はあまりに速く終わるため目にとまらないほどです。速度を視覚的に確認できるよう、アニメーションでは同じクエリを繰り返し実行しています。
* 実行時間: **33 ms**
* 処理行数: **3,045 万行**
* スループット: **毎秒 9 億 2,120 万行**、**4.61 GB/s**
> 人間が[まばたき](https://www.ucl.ac.uk/news/2005/jul/blink-and-you-miss-it)をする時間は約 100〜150 ms です。33 ms という処理速度は、まばたきのおよそ **4 分の 1** であり、[カメラのシャッター音](https://www.davemorrowphotography.com/shutter-speed-chart?utm_source=chatgpt.com#Shutter_Speed_Chart_Camera_Technique_Video)が鳴るほどの速さです。
### スケールアップ: 数十億行が新たな標準に
現在のデータセットは、数十億、数兆、さらには数千兆行に達することも珍しくありません。[Tesla は負荷テストのために、ClickHouse に 1,000 兆行を超えるデータを取り込みました](https://clickhouse.com/ja/blog/how-tesla-built-quadrillion-scale-observability-platform-on-clickhouse-jp)。
それでは、スケールさせてみましょう。
### 10 億行: まばたき 2 回分の時間
同じく 89 コアの 1 ノード上で、**10 億行**に対して**同じクエリ**を実行しました。
USE uk_b1; -- 1B rows dataset
SELECT
town,
formatReadableQuantity(sum(price)) AS total_revenue
FROM uk_price_paid
GROUP BY town
ORDER BY sum(price) DESC
LIMIT 3
SETTINGS
enable_parallel_replicas = false;
* 実行時間: **207 ms**
* 処理行数: **10 億行**
* スループット: **毎秒 48 億 3,000 万行**、**29.01 GB/s**
> 2 回まばたきをするほどの時間で、10 億行のグループ化が完了しました。
### 1,000 億行: 指を鳴らすほどの時間
今度は **1,000 億行**に引き上げ、同じクエリを再度実行します (89 コアの 1 ノード上)。
USE uk_b100; -- 100B rows dataset
SELECT
town,
formatReadableQuantity(sum(price)) AS total_revenue
FROM uk_price_paid
GROUP BY town
ORDER BY sum(price) DESC
LIMIT 3
SETTINGS
enable_parallel_replicas = false;
┌─town───────┬─total_revenue────┐
│ LONDON │ 3.92 quadrillion │
│ BRISTOL │ 359.61 trillion │
│ MANCHESTER │ 266.83 trillion │
└────────────┴──────────────────┘
3 rows in set. Elapsed: 16.581 sec. Processed 100.00 billion rows, 600.00 GB (6.03 billion rows/s., 36.19 GB/s.)
Peak memory usage: 572.19 MiB.
1,000 億行の処理が約 17 秒で完了しました 🐌。
スループットは良好ですが (毎秒 60 億 3,000 万行、36.19 GB/s)、アニメーションのループとしてここで見せるには実行時間が長すぎます。ターボスイッチを入れる時間です 🏎️。
**parallel replicas を有効化する**と、同じクエリがサービス内の**全ノードの全コア**へとファンアウトされます。
USE uk_b100; -- 100B rows dataset
SELECT
town,
formatReadableQuantity(sum(price)) AS total_revenue
FROM uk_price_paid
GROUP BY town
ORDER BY sum(price) DESC
LIMIT 3
SETTINGS
enable_parallel_replicas = true;
* 実行時間: **414 ms**
* 処理行数: **1,000 億行**
* スループット: **毎秒 2,418 億 3,000 万行**、**1.45 TB/s**
> **414 ms** は[指をパチンと鳴らす](https://en.wikipedia.org/wiki/Finger_snapping)か、1 回手を叩くほどの時間です。その瞬間に、ClickHouse は **1,000 億行**を**毎秒 1 テラバイト以上**の速度で集計しました。
まさに圧巻です。1,000 億行を 1 秒未満、TB/s 級のスループットで集計できました。
では、ClickHouse はこれをどのように実現しているのでしょうか。単一ノード上で ClickHouse が GROUP BY をこれほど高速化できる仕組みから順に、内部構造を解き明かしていきましょう。
## ClickHouse が GROUP BY を高速化する仕組み
ClickHouse が高速なのは、[クエリパイプライン](https://clickhouse.com/docs/ja/guides/clickhouse/performance-and-monitoring/understanding-query-execution-with-the-analyzer#query-pipeline) (物理クエリプラン) 内の**並列ストリーム** (*高速道路の車線をイメージしてください*) を利用し、すべての CPU コアにクエリを並列分散させて**部分集計状態 (partial aggregation states)** を構築するためです。
### 列指向ストレージによる下地作り
ClickHouse は[列指向ストレージ](https://clickhouse.com/docs/ja/concepts/core-concepts/parts)を採用しています。クエリは参照されたカラムのみを読み取り、カラムごとの高い圧縮率によってスキャンバイト数を抑え、連続したデータに対してベクトル化 (SIMD) 演算を実行します。その結果、I/O が低く抑えられ演算密度が高くなり、CPU コアごとに 1 つの並列クエリパイプラインストリームを実行するのに最適な環境が整います。
### 1 コアあたり 1 つのストリーム
ClickHouse は、**CPU コアごとに 1 つの並列クエリパイプラインストリーム**を実行します。各ストリームは行のスキャン、フィルターの適用、部分集計状態の構築を並列で行います。
クエリエンジン内部では次のような処理が行われます。
**① 多数のパート範囲を同時に読み込み**
ClickHouse は、複数の[データパート](https://clickhouse.com/docs/ja/concepts/core-concepts/parts)範囲 ([インデックス分析](https://clickhouse.com/docs/ja/concepts/core-concepts/primary-indexes)によって選択された連続する行ブロック) から異なるストリームへとデータを同時にスキャンします。
**② 並列フィルタリングと集計**
各ストリームは互いに重複しない専用の範囲を担当し、[ベクトル化実行](https://clickhouse.com/blog/cpu-dispatch-in-clickhouse) (SIMD) を使用して行のフィルタリングと部分状態の更新を行います。
**③ 部分状態のマージ**
すべてのストリームからの部分状態が統合され、最終的な結果になります。このマージステップは通常、**複数のマージスレッドにまたがって並列に実行されます**。ただし、従来は必ずしも並列化されていなかった特殊な GROUP BY バリアントも存在します。[私たちはこれらの実装に対しても完全並列マージをサポートするよう拡張を続けています](/blog/clickhouse-release-25-11#parallel-merge-for-small-group-by)。
*(これとは別に、クエリが COUNT、SUM、MAX などの**複数の集計**を計算する場合、各集計は独自の部分状態を保持し、それらの**マージ処理は互いに並行して実行されます**。)*
**④ ソートと LIMIT の適用**
マージされた結果が並べ替えられ、クエリに LIMIT 句がある場合はそれが適用されます。
**⑤ 応答の返却**
最終結果がクライアントに返送されます。
この設計が機能するのは、パイプラインストリームが最終的な値を直接計算しないためです。後から 1 つの正しい結果へと結合できる**マージ可能な部分状態**を構築するのです。
### 部分状態が重要な理由
> 部分状態があるからこそ、並列化が可能になります。
*これがなければ*、後から GROUP BY をコア間やノード間に効率的に分割することはできません。すべてのコア (またはノード) が、特定のグループに属するすべての行を参照しなければならなくなるからです。*部分状態があれば*、どのコア (またはノード) でも任意の行のサブセットを独立して処理し、1 つの正しい結果に結合できるマージ可能な状態を出力できます。
この柔軟性により、以下が実現します。
* **効率的で明快なパイプラインストリームのスケジューリング**: ストリームはキーごとに事前にパーティショニングされたデータを必要とせず、任意の範囲をスキャンできます。
* **動的な負荷分散**: あるストリームでデータの偏り (スキュー) が発生した場合でも、[行を他のストリームに再ルーティング](https://clickhouse.com/docs/ja/concepts/core-concepts/academic-overview#4-2-multi-core-parallelization)できるため、全体のスループットを高く維持できます。重要なのは、マージ可能な状態を生成することだけです。
ClickHouse は、[170 以上のすべての集計関数](https://sql.clickhouse.com/?query=U0VMRUNUICogCkZST00gc3lzdGVtLmZ1bmN0aW9ucwpXSEVSRSBpc19hZ2dyZWdhdGUKQU5EIGVtcHR5KGFsaWFzX3RvKQpPUkRFUiBCWSBuYW1l&chart=eyJ0eXBlIjoibGluZSIsImNvbmZpZyI6eyJ4YXhpcyI6InllYXIiLCJ5YXhpcyI6Ijk5dGhfYXZnX3RlbXAiLCJzZXJpZXMiOiJjb3VudHJ5IiwidGl0bGUiOiJUZW1wZXJhdHVyZSBieSBjb3VudHJ5IGFuZCB5ZWFyIn19&run_query=true&tab=results) (およびその[コンビネータ](https://www.youtube.com/watch?v=7ApwD0cfAFI)) をコア間で並列化し、後述するようにノード間でも並列化します。
次に、クエリパイプラインの集計ステージの内部について、さらに詳しく見ていきましょう。
### GROUP BY 実行パイプラインの内部
GROUP BY クエリは、各パイプラインストリームにおいて*ハッシュ集計アルゴリズム*を用いて独立して処理されます。各ストリームはインメモリハッシュテーブルを保持し、各キー (例: town) が部分集計状態を指し示します。
> **GROUP BY を支えるハッシュテーブル**
USE uk_b10; -- 10B rows dataset
SELECT
county,
town,
district,
formatReadableQuantity(count()) AS properties_sold,
formatReadableQuantity(sum(price)) AS total_sales_value,
formatReadableQuantity(avg(price)) AS average_sale_price
FROM uk_price_paid
GROUP BY
county,
town,
district
ORDER BY sum(price) DESC
LIMIT 10
SETTINGS
enable_parallel_replicas = false;
**最大スペックの 89 コアノード**において、実行時間は **14.4 秒** (毎秒約 6 億 9,200 万行) でした。まずまずの速度ですが、インタラクティブとは言えません。ダッシュボードやアドホック分析において 14 秒という待ち時間は長く感じられます。
> 靴紐を結ぶ、赤信号を待つ、あるいは電子レンジでピザを温め直すくらいの時間です (イタリアの読者の皆さん、ごめんなさい)。
このようにカーディナリティが固定されているワークロード (英国で新しい州や町が次々と増えるわけではありません) では、[インクリメンタルなマテリアライズドビュー](https://clickhouse.com/docs/ja/concepts/features/materialized-views/incremental-materialized-view)を利用するのが賢明な解決策です。地理情報ごとに事前集計しておくことでデータ量が大幅に削減され、ミリ秒単位の応答時間が[保証](https://clickhouse.com/blog/accelerating-clickhouse-json-queries-for-fast-bluesky-dashboards#final-takeaways-for-clickhouse-users)されます。
> **マテリアライズドビュー: 部分状態をディスクに保存する**
USE uk_b10; -- 10B rows dataset
SELECT
county,
toStartOfQuarter(date) AS qtr,
uniq(street) AS distinct_streets_with_sales
FROM
uk_price_paid
GROUP BY
county, qtr
ORDER BY
qtr, county;
クエリ自体はシンプルですが、どの集計関数を選択するかが鍵となります。
*注: このベンチマークは、異なる集計関数によって生成される部分状態のサイズの違いに焦点を当てています。これまでのベンチマークとは異なる新しい GROUP BY クエリを使用しているため、**89 コアのノードではなく 16 コアのノードで実行**しており、コア数に関係なく parallel replicas が効率的にスケールすることも同時に示しています。*
### 軽量なケース: uniq
uniq 集計関数は近似値を算出しますが、非常に効率的です。アダプティブサンプリング (最大 65,536 ハッシュ) を使用し、状態をコンパクトに保ちます。
*完全な結果は[こちら](https://github.com/ClickHouse/examples/blob/9884ee05fbd20223cd9c67ab736dfdbc1024c133/blog-examples/ParallelReplicasBench/results/2025-09-02_horiz_uniq_uk_b10_n1-3-10-20-40-80-100_c16_r10/summary_matrix-nodes-1_3_10_20_40_80_100_cores-16_2025-09-02_050856.json)で確認できます。*
* **ノードあたりの送信状態サイズ:** *送信ノード*あたり **~58 MiB** (3 ノード構成時の送信側 2 ノード) から、100 ノード構成時のノードあたり **~4.2 MiB** まで減少
* **コーディネーターの受信総量:** 3 ノード構成時の **~115 MiB** (2×58 MiB) から、100 ノード構成時の **~416 MiB** (99×4.2 MiB)
→ コーディネーターへのファンインは小さく保たれ、ノードが増えるにつれてノードあたりのトラフィックは縮小します。
### 重いケース: uniqExact
uniqExact 集計関数は正確な COUNT DISTINCT を計算します。コーディネーターがマージする前に、各ノードが検出したすべての個別値を型に応じて生の値またはハッシュとして追跡・保持しなければならないため、部分状態が大きくなります。
*完全な結果は[こちら](https://github.com/ClickHouse/examples/blob/9884ee05fbd20223cd9c67ab736dfdbc1024c133/blog-examples/ParallelReplicasBench/results/2025-09-02_horiz_uniqExact_uk_b10_n1-3-10-20-40-80-100_c16_r10/summary_matrix-nodes-1_3_10_20_40_80_100_cores-16_2025-09-02_055212.json)にあります。*
* **ノードあたりの送信状態サイズ:** *送信ノード*あたり **~225 MiB** (3 ノード構成時の送信側 2 ノード) から、100 ノード構成時のノードあたり **~16 MiB** まで減少
* **コーディネーターの受信総量:** 3 ノード構成時の **~450 MiB** (2×225 MiB) から、100 ノード構成時の **~1.5 GiB** (99×16 MiB)
状態のサイズが uniq と比較して **~4 倍大きく**なるため、実行時間は短縮されるものの、ネットワークファンインとマージ処理の負荷がより顕著になります。
ただし、これは一概には言えません。uniqExact の状態サイズは**カーディナリティ**に応じてスケールするためです。このデータでは、(county, quarter) ごとの個別の通りの数は最大約 1 万 7,000 件で安定しており、この例での負荷は比較的穏やかな部類に入ります。
> **カーディナリティに関する注意点:** 個別値の数が爆発的に増加する場合 (例: 数十億のユーザーに対する `uniqExact(user_id)` や Web 全体を対象とした `uniqExact(url)`)、状態は `uniq` と比較して**数桁大きくなる**可能性があり、ネットワークオーバーヘッドもそれに応じて増大します。
### 個別値カウントに関するまとめ
parallel replicas は 170 種類以上の集計関数すべてをスケールさせますが、その効率は部分状態のサイズに左右されます。
* **軽量な状態** (*sum*、*count*、*uniq* など): 非常に良好にスケールします。
* **重い状態** (*uniqExact*): 同様にスケールしますが、ネットワークおよびマージのコストが高くなります。
* **その他の選択肢** (*uniqHLL12*): 1 万〜1 億件の個別値に対して約 1〜2% の誤差で約 2.5 KB の固定状態サイズを維持します。ごく小さなデータセットでは精度が落ち、1 億件を超えると劣化しますが、一部のワークロードにとっては実用的なバランスの取れた選択肢となります。
重い集計であってもスケールすることに変わりはありませんが、部分状態が大きくなるとネットワークトラフィックとマージコストが増加します。そのトレードオフが見合うかどうかは、各ノードが処理すべきデータ量に依存します。このことが、parallel replicas の効率性を保つために ClickHouse が適用している安全策の話題へとつながっていきます。
## スケーリングの限界と安全策
クエリは単にノード数を増やせば速くなるというものでもありません。
**小規模なデータセット (例: 100 億行)** では、parallel replicas はノード数が少ないうちは良好にスケールしますが、ノード数が増えると各ノードの処理対象データが枯渇し、コーディネーションのオーバーヘッドが支配的になります。
**大規模なデータセット (例: 1 兆行)** では逆のことが起こります。ノード数が少ないと負荷をさばききれませんが、ノード数を増やすと各ノードに十分な処理対象行が残るため、コーディネーションのコストに見合う形で綺麗にスケールします。
ClickHouse は、parallel replicas の効率性を保つために 2 つの安全策を設けています。
* [parallel_replicas_min_number_of_rows_per_replica](https://clickhouse.com/docs/ja/reference/settings/session-settings) — ノードにオーバーヘッドを正当化できるだけの十分な行数が割り当てられる場合にのみ、そのノードを parallel replica として稼働させます。
* [max_parallel_replicas](https://clickhouse.com/docs/ja/reference/settings/session-settings) — 1 つのクエリに対して parallel replica として使用されるノード数の上限を設定します (デフォルトは 1000)。
内部では、クエリプランナーがクエリによって実際にスキャンおよび処理されると予測される行数である **rows_to_read** を (インデックス分析などを通じて) 見積もります。上記 2 つの設定値が組み合わさることで、parallel replicas を使用する価値があるかどうか、使用する場合はノード数をいくつにするかが判断されます。
* **有効化の判定ルール**: クエリに対して parallel replicas が有効化されるのは、次の条件を満たす場合のみです。
rows_to_read >= 2* parallel_replicas_min_number_of_rows_per_replica
rows_to_read / parallel_replicas_min_number_of_rows_per_replica
(ただし min(クラスターサイズ, max_parallel_replicas) を上限とする)