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

テーブルのシャードとレプリカ


注記

このトピックは ClickHouse Cloud には適用されません。ClickHouse Cloud では、Parallel Replicas は従来のシェアードナッシング型 ClickHouse クラスターにおける複数シャードのように機能し、オブジェクトストレージがレプリカを代替することで、高可用性とフォールトトレランスを実現します。

ClickHouse におけるテーブルシャードとは？​

従来型の shared-nothing アーキテクチャの ClickHouse クラスターでは、① データ量が単一サーバーで扱うには大きすぎる場合、または ② 単一サーバーではデータ処理が遅すぎる場合にシャーディングが行われます。次の図はケース ①、つまり uk_price_paid_simple テーブルが 1 台のマシンの容量を超えている状況を示しています：

SHARDS

このような場合、データはテーブルシャードという形で複数の ClickHouse サーバーに分割して配置できます：

SHARDS

各 ^^shard^^ はデータの一部を保持し、独立してクエリ可能な通常の ClickHouse テーブルとして動作します。ただし、そのシャードに対するクエリで処理されるのは、そのシャードに含まれるサブセットのみであり、これはデータ分布によっては妥当なユースケースになり得ます。一般的には、（多くの場合サーバーごとに）distributed table が全データセットに対する統一的なビューを提供します。この distributed table 自体はデータを保持せず、すべてのシャードに SELECT クエリを転送して結果を集約し、INSERT をルーティングしてデータを均等に分散させます。

分散テーブルの作成​

SELECT クエリのフォワーディングと INSERT のルーティングを説明するために、2 台の ClickHouse サーバー上の 2 つのシャードに分割された、テーブルパーツとは で使用したサンプルテーブルを用います。まず、この構成に対応する ^^Distributed table^^ を作成するための DDL ステートメントを示します。

CREATE TABLE uk.uk_price_paid_simple_dist ON CLUSTER test_cluster
(
    date Date,
    town LowCardinality(String),
    street LowCardinality(String),
    price UInt32
)
ENGINE = Distributed('test_cluster', 'uk', 'uk_price_paid_simple', rand())

ON CLUSTER 句は DDL ステートメントを分散 DDL ステートメントにし、ClickHouse に対して、test_cluster のクラスター定義に記載されているすべてのサーバー上にテーブルを作成するよう指示します。分散 DDL を利用するには、クラスターアーキテクチャ内に追加の Keeper コンポーネントが必要です。

Distributed エンジンのパラメータとして、^^cluster^^ 名 (test_cluster)、シャードされたターゲットテーブルのデータベース名 (uk)、そのターゲットテーブル名 (uk_price_paid_simple)、および INSERT ルーティング用のシャーディングキーを指定します。この例では、rand 関数を使用して、行を各シャードにランダムに割り当てます。ただし、ユースケースに応じて、複雑なものも含め任意の式をシャーディングキーとして利用できます。次のセクションでは、INSERT ルーティングがどのように動作するかを説明します。

INSERT ルーティング​

次の図は、ClickHouse において ^^分散テーブル^^ への INSERT がどのように処理されるかを示しています。

シャード

① ^^分散テーブル^^ を対象とする INSERT（1 行のみ）が、テーブルをホストしている ClickHouse サーバーに、直接またはロードバランサー経由で送信されます。

② INSERT の各行（この例では 1 行）に対して、ClickHouse はシャーディングキー（ここでは rand()）を評価し、その結果を ^^シャード^^ サーバーの台数で割った余りを取り、その値をターゲットサーバーIDとして使用します（ID は 0 から始まり 1 ずつ増加します）。その後、その行は該当するサーバー上のテーブルの ^^シャード^^ に転送され、③ 挿入されます。

次のセクションでは、SELECT フォワーディングの仕組みについて説明します。

SELECT のフォワーディング​

この図は、ClickHouse で ^^distributed table^^ を使用した場合に、SELECT クエリがどのように処理されるかを示しています。

SHARDS

① ^^distributed table^^ を対象とする SELECT 集約クエリが、対応する ClickHouse サーバーに送信されます（直接、またはロードバランサー経由）。

② ^^distributed table^^ は、その対象テーブルのシャードを保持しているすべてのサーバーにクエリを転送し、各 ClickHouse サーバーはローカルの集約結果を 並列に 計算します。

その後、最初に ^^distributed table^^ が配置されている ClickHouse サーバーが、③ すべてのローカル結果を収集し、④ それらをマージして最終的なグローバル結果を作成し、⑤ クエリ送信元に返します。

ClickHouse におけるテーブルレプリカとは？​

ClickHouse のレプリケーションは、複数サーバー間で ^^shard^^ データのコピー を保持することで、データ整合性 と フェイルオーバー を実現します。ハードウェア障害は避けられないため、各 ^^shard^^ に複数のレプリカを持たせることで、レプリケーションによってデータ損失を防ぎます。書き込みは任意の ^^replica^^ に対して、直接、または操作対象の ^^replica^^ を選択する distributed table 経由で行うことができます。変更は他のレプリカへ自動的に伝播されます。障害やメンテナンスが発生した場合でも、データは他のレプリカ上で利用可能な状態が保たれ、障害ホストが復旧すると自動的に同期され、最新状態が維持されます。

レプリケーションには、クラスタアーキテクチャ 内の Keeper コンポーネントが必要である点に注意してください。

次の図は 6 台のサーバーからなる ClickHouse ^^cluster^^ を表しており、前述の 2 つのテーブルシャード Shard-1 と Shard-2 がそれぞれ 3 つのレプリカを持っています。この ^^cluster^^ に対してクエリが送信されます。

SHARDS

クエリ処理はレプリカがない構成と同様に動作し、各 ^^shard^^ からは 1 つの ^^replica^^ だけがクエリを実行します。

レプリカはデータ整合性とフェイルオーバーを保証するだけでなく、異なるレプリカ間で複数のクエリを並列実行できるようにすることで、クエリ処理のスループットも向上させます。

① ^^distributed table^^ を対象とするクエリが、直接またはロードバランサー経由で対応する ClickHouse サーバーに送信されます。

② ^^Distributed table^^ はクエリを各 ^^shard^^ から 1 つの ^^replica^^ に転送し、選択された ^^replica^^ をホストする各 ClickHouse サーバーがローカルクエリ結果を並列に計算します。

残りの処理は、レプリカがない構成の場合と同様であり、上の図には示していません。最初に対象となった ^^distributed table^^ をホストする ClickHouse サーバーがすべてのローカル結果を収集し、それらをマージして最終的なグローバル結果を生成し、クエリ送信元に返します。

なお、② におけるクエリ転送戦略は ClickHouse で設定可能です。デフォルトでは（上の図とは異なり）、^^distributed table^^ は利用可能であればローカルの ^^replica^^ を優先しますが、他の負荷分散戦略も使用できます。

さらに詳しい情報を得るには​

テーブルのシャードとレプリカについてのこの高レベルな概要を踏まえ、さらに詳しい情報が必要な場合は、デプロイとスケーリングガイドを参照してください。

また、ClickHouse のシャードとレプリカをより深く理解するには、次のチュートリアル動画も強く推奨します。