ClickHouseスケーリング、移行管理、PHPからクラスターへのリクエスト送信

前の記事で、 media2社でClickHouse DBMSを実装および使用した経験を共有しました。 現在の記事では、同じ物理サーバー内でデータを保存および処理できなくなったときに分析されるデータ量と負荷の増加に伴って発生するスケーリングの問題に対処します。 また、 DDLクエリをClickHouseクラスターに移行するために開発したツールについても説明します。

ClickHouseは 、異なるデータセンターにあるクラスターで動作するように特別に設計されました。 DBMSは、数百のノードに線形にスケーリングされます。 したがって、たとえば、執筆時点でのYandex.Metricaは、400以上のノードのクラスターです。

ClickHouseは、すぐに使用できるシャーディングとレプリケーションを提供し、テーブルごとに個別に柔軟に構成できます。 レプリケーションにはApache ZooKeeperが必要です（バージョン3.4.5以降を推奨）。 信頼性を高めるために、5ノードのZKクラスター（アンサンブル）を使用します。 クォーラムを提供するには、奇数のZKノード（3または5など）を選択する必要があります。 ZKはSELECT操作では使用されませんが、たとえば、ALTERクエリで使用されて列を変更し、各レプリカの指示を保存することに注意してください。

シャーディング

ClickHouseのシャーディングを使用すると、クラスター内のすべてのノードに分散してクラスター内のデータのチャンクを書き込みおよび保存し、クラスター内のすべてのノードでデータを並列処理（読み取り）して、スループットを向上させ、待ち時間を短縮できます。 たとえば、GROUP BY ClickHouseを使用したクエリでは、リモートノードで集約が実行され、集約関数の中間状態がリクエストの開始ノードに渡され、そこで集約されます。

シャーディングには、データを格納しない特別な分散エンジンが使用されますが、SELECTクエリをシャードテーブル（データを含むテーブル）に委任し、受信データの後続処理を行います。 シャードへのデータの書き込みは、2つのモードで実行できます。1）分散テーブルとオプションのシャーディングキーを使用するか、2）シャードテーブルに直接入力し、そこから分散テーブルを介してデータを読み取ります。 これらのモードをさらに詳しく考えてみましょう。

最初のモードでは、データはシャードキーを使用して分散テーブルに書き込まれます。 最も単純なケースでは、ランダム化キーは乱数、つまりrand（）関数を呼び出した結果である場合があります。 ただし、ハッシュ関数の値をテーブルのフィールドからシャーディングキーとして取得することをお勧めします。これにより、一方では小さなデータセットを一方のシャードにローカライズでき、他方ではクラスター内の異なるシャードでそのようなセットをかなり均等に分散できます。 たとえば、ユーザーのセッション識別子（sess_id）を使用すると、同じシャードの1人のユーザーにページ表示をローカライズできますが、異なるユーザーのセッションはクラスター内のすべてのシャードに均等に分散されます（sess_idフィールドの値に適切な分布がある場合）。 シャーディングキーは、非数値または複合にすることもできます。 この場合、組み込みのハッシュ関数cityHash64を使用できます。 考慮モードでは、クラスターノードの1つに書き込まれたデータは、シャーディングキーを使用して必要なシャードに自動的にリダイレクトされますが、トラフィックが増加します。

より複雑な方法は、ClickHouseの外部から必要なシャードを計算し、シャードテーブルに直接書き込むことです。 ここでの困難は、利用可能なノードシャードのセットを知る必要があるという事実によるものです。 ただし、この場合、記録はより効率的になり、シャーディングメカニズム（目的のシャードを決定する）はより柔軟になります。

複製

ClickHouseはデータ複製をサポートし、レプリカのデータ整合性を確保します。 データ複製には、MergeTreeファミリーの特別なエンジンが使用されます。

• ReplicatedMergeTree
• ReplicatedCollapsingMergeTree
• ReplicatedAggregatingMergeTree
• ReplicatedSummingMergeTree

多くの場合、レプリケーションはシャーディングと組み合わせて使用​​されます。 たとえば、6ノードのクラスターには、2つのレプリカを持つ3つのシャードが含まれる場合があります。 レプリケーションはシャーディングメカニズムに依存せず、個々のテーブルのレベルで機能することに注意してください。

データはどのレプリカテーブルにも書き込むことができ、ClickHouseはすべてのレプリカ間でデータを自動的に同期します。

ClickHouseクラスターの構成例

例として、4つのノードch63.smi2, ch64.smi2, ch65.smi2, ch66.smi2さまざまな構成を検討します。 設定は、構成ファイル/etc/clickhouse-server/config.xmlに含まれています。

1つの断片と4つのレプリカ

 <remote_servers> <!-- One shard, four replicas --> <repikator> <shard> <!-- replica 01_01 --> <replica> <host>ch63.smi2</host> </replica> <!-- replica 01_02 --> <replica> <host>ch64.smi2</host> </replica> <!-- replica 01_03 --> <replica> <host>ch65.smi2</host> </replica> <!-- replica 01_04 --> <replica> <host>ch66.smi2</host> </replica> </shard> </repikator> </remote_servers> 

テーブル作成スキームの例：

指定された構成のテーブルを作成するSQLクエリの例：

 CREATE DATABASE IF NOT EXISTS dbrepikator ; CREATE TABLE IF NOT EXISTS dbrepikator.anysumming_repl_sharded ( event_date Date DEFAULT toDate(event_time), event_time DateTime DEFAULT now(), body_id Int32, views Int32 ) ENGINE = ReplicatedSummingMergeTree('/clickhouse/tables/{repikator_replica}/dbrepikator/anysumming_repl_sharded', '{replica}', event_date, (event_date, event_time, body_id), 8192) ; CREATE TABLE IF NOT EXISTS dbrepikator.anysumming_repl AS dbrepikator.anysumming_repl_sharded ENGINE = Distributed( repikator, dbrepikator, anysumming_repl_sharded , rand() ) 

この構成の利点：

• データを保存する最も信頼できる方法。

短所：

• ほとんどのタスクでは、データの余分なコピーが保存されます。
• この構成にはシャードが1つしかないため、異なるノードでSELECTクエリを並行して実行することはできません。
• すべてのノード間でデータを繰り返し複製するには、追加のリソースが必要です。

1つのレプリカに4つの破片

 <remote_servers> <!-- Four shards, one replica --> <sharovara> <!-- shard 01 --> <shard> <!-- replica 01_01 --> <replica> <host>ch63.smi2</host> </replica> </shard> <!-- shard 02 --> <shard> <!-- replica 02_01 --> <replica> <host>ch64.smi2</host> </replica> </shard> <!-- shard 03 --> <shard> <!-- replica 03_01 --> <replica> <host>ch65.smi2</host> </replica> </shard> <!-- shard 04 --> <shard> <!-- replica 04_01 --> <replica> <host>ch66.smi2</host> </replica> </shard> </sharovara> </remote_servers> 

指定された構成のテーブルを作成するSQLクエリの例：

 CREATE DATABASE IF NOT EXISTS testshara ; CREATE TABLE IF NOT EXISTS testshara.anysumming_sharded ( event_date Date DEFAULT toDate(event_time), event_time DateTime DEFAULT now(), body_id Int32, views Int32 ) ENGINE = ReplicatedSummingMergeTree('/clickhouse/tables/{sharovara_replica}/sharovara/anysumming_sharded_sharded', '{replica}', event_date, (event_date, event_time, body_id), 8192) ; CREATE TABLE IF NOT EXISTS testshara.anysumming AS testshara.anysumming_sharded ENGINE = Distributed( sharovara, testshara, anysumming_sharded , rand() ) 

この構成の利点：

• この構成には4つのシャードがあるため、クラスターのすべてのノードでSELECTクエリを同時に並行して実行できます。

欠点：

• データを保存する最も信頼性の低い方法（ノードの損失は、データの一部の損失につながります）。

2つの断片、2つのレプリカ

 <remote_servers> <!-- Two shards, two replica --> <pulse> <!-- shard 01 --> <shard> <!-- replica 01_01 --> <replica> <host>ch63.smi2</host> </replica> <!-- replica 01_02 --> <replica> <host>ch64.smi2</host> </replica> </shard> <!-- shard 02 --> <shard> <!-- replica 02_01 --> <replica> <host>ch65.smi2</host> </replica> <!-- replica 02_02 --> <replica> <host>ch66.smi2</host> </replica> </shard> </pulse> </remote_servers> 

指定された構成のテーブルを作成するSQLクエリの例：

 CREATE DATABASE IF NOT EXISTS dbpulse ; CREATE TABLE IF NOT EXISTS dbpulse.normal_summing_sharded ( event_date Date DEFAULT toDate(event_time), event_time DateTime DEFAULT now(), body_id Int32, views Int32 ) ENGINE = ReplicatedSummingMergeTree('/clickhouse/tables/{pulse_replica}/pulse/normal_summing_sharded', '{replica}', event_date, (event_date, event_time, body_id), 8192) ; CREATE TABLE IF NOT EXISTS dbpulse.normal_summing AS dbpulse.normal_summing_sharded ENGINE = Distributed( pulse, dbpulse, normal_summing_sharded , rand() ) 

この構成は、最初と2番目の例から最高の品質を実現します。

• この構成には2つのシャードがあるため、クラスター内の各シャードでSELECTクエリを並行して実行できます。
• データを保存する比較的信頼性の高い方法（1つのクラスターノードが失われても、データの一部が失われることはありません）。

Ansibleクラスターの構成例

ansibleのクラスター構成は次のようになります。

 - name: "pulse" shards: - { name: "01", replicas: ["ch63.smi2", "ch64.smi2"]} - { name: "02", replicas: ["ch65.smi2", "ch66.smi2"]} - name: "sharovara" shards: - { name: "01", replicas: ["ch63.smi2"]} - { name: "02", replicas: ["ch64.smi2"]} - { name: "03", replicas: ["ch65.smi2"]} - { name: "04", replicas: ["ch66.smi2"]} - name: "repikator" shards: - { name: "01", replicas: ["ch63.smi2", "ch64.smi2","ch65.smi2", "ch66.smi2"]} 

ClickHouseクラスターを操作するためのPHPドライバー

前の記事で、 ClickHouse用のオープンソースPHPドライバーについて既に説明しました。

ノードの数が多くなると、クラスター管理が不便になります。 したがって、DDLクエリをClickHouseクラスターに移行するためのシンプルで十分に機能的なツールを開発しました。 次に、その機能の例を簡単に説明します。

クラスターに接続するには、 ClickHouseDB\Clusterクラスが使用されます。

 $cl = new ClickHouseDB\Cluster( ['host'=>'allclickhouse.smi2','port'=>'8123','username'=>'x','password'=>'x'] ); 

allclickhouse.smi2 DNSレコードには、すべてのノードのIPアドレスがリストされています： ch63.smi2, ch64.smi2, ch65.smi2, ch66.smi2 。これにより、 ラウンドロビンDNSメカニズムを使用できます。

ドライバーはクラスターに接続し、DNSレコードにリストされている各ノードにping要求を送信します。

すべてのクラスターノードへの最大接続時間の設定は、次のように構成されます。

 $cl->setScanTimeOut(2.5); // 2500 ms 

クラスタレプリカのステータスの確認は、次のように実行されます。

 if (!$cl->isReplicasIsOk()) { throw new Exception('Replica state is bad , error='.$cl->getError()); } 

ClickHouseクラスターの状態は、次のように確認されます。

• DNSレコードにリストされているすべてのクラスターノードへの接続が確認されます。
• SQLクエリが各ノードに送信されるため、ClickHouseクラスターのすべてのレプリカのステータスを確認できます。

ZKクラスターへのクエリが実行されるデータを受信するときにlog_max_index, log_pointer, total_replicas, active_replicasの値を減算しない場合、クエリの実行速度を上げることができます。

軽量検証の場合、ドライバーに特別なフラグを設定する必要があります。

 $cl->setSoftCheck(true); 

使用可能なすべてのクラスターのリストを取得するには、次のようにします。

 print_r($cl->getClusterList()); // result // [0] => pulse // [1] => repikator // [2] => sharovara 

たとえば、上記で説明したクラスターの構成を取得するには、次のようにします。

 foreach (['pulse','repikator','sharovara'] as $name) { print_r($cl->getClusterNodes($name)); echo "> $name , count shard = ".$cl->getClusterCountShard($name)." ; count replica = ".$cl->getClusterCountReplica($name)."\n"; } //: //> pulse , count shard = 2 ; count replica = 2 //> repikator , count shard = 1 ; count replica = 4 //> sharovara , count shard = 4 ; count replica = 1 

クラスター名またはシャードテーブルからノードのリストを取得する：

 $nodes=$cl->getNodesByTable('sharovara.body_views_sharded'); $nodes=$cl->getClusterNodes('sharovara'); 

クラスターの各ノードにリクエストを送信して、テーブルのサイズまたはすべてのテーブルのサイズを取得します。

 foreach ($nodes as $node) { echo "$node > \n"; print_r($cl->client($node)->tableSize('test_sharded')); print_r($cl->client($node)->tablesSize()); } //    $cl->getSizeTable('dbName.tableName'); 

クラスタテーブルのリストを取得します。

 $cl->getTables() 

クラスター内のリーダーの定義：

 $cl->getMasterNodeForTable('dbName.tableName') //     is_leader=1 

たとえば、構造の削除または変更に関連するリクエストは、 is_leaderフラグがis_leaderれたノードに送信is_leaderます。

クラスター内のテーブルのデータを消去する：

 $cl->truncateTable('dbName.tableName')` 

DDLクエリ移行ツール

当社は、 MyBatis Migrationsを使用して、リレーショナルDBMSのDDLクエリを移行します。

Habréの移行ツールについては、すでに次のように書いています。

• データベース構造のバージョン化された移行：主要なアプローチ
• PHPixie Migrateを使用した簡単な移行
• 移行スクリプトまたはMyBatis Scheme Migration Extendedの管理

ClickHouseクラスターを使用するには、同様のツールが必要でした。

執筆時点では、ClickHouseにはDDLクエリに関連付けられた多くの機能（制限）があります。 引用 ：

INSERT、ALTERが複製されます（詳細については、ALTERクエリの説明を参照してください）。 クエリテキストではなく、圧縮データが複製されます。 CREATE、DROP、ATTACH、DETACH、RENAMEリクエストはレプリケートされません。つまり、同じサーバーを参照します。 CREATE TABLEクエリは、クエリが実行されるサーバーに新しいレプリケートテーブルを作成します。 また、他のサーバーにそのようなテーブルが既にある場合、新しいレプリカを追加します。 DROP TABLEクエリは、クエリが実行されるサーバー上にあるレプリカを削除します。 RENAMEクエリは、レプリカの1つにあるテーブルの名前を変更します。つまり、異なるレプリカにある複製されたテーブルには、異なる名前を付けることができます。

ClickHouse開発チームはこの方向での作業を既に発表していますが、現在は外部ツールでこの問題を解決する必要があります。 DDLクエリをClickHouseクラスターに移行するためのphpMigrationsClickhouseツールの簡単なプロトタイプを作成しました。 そして、私たちの計画はphpMigrationsClickhouseをPHP言語から抽象化することです。

他のプログラミング言語で実装できるphpMigrationsClickhouseで現在使用されているアルゴリズムについて説明します。

現在、 phpMigrationsClickhouseの移行手順は以下で構成されています。

• エラーが発生した場合にロールバックおよびロールバックする必要があるSQLクエリ。
• SQLクエリを実行するクラスターの名前。

次のコードを含むPHPファイルを作成します。

 $cluster_name = 'pulse'; $mclq = new \ClickHouseDB\Cluster\Migration($cluster_name); $mclq->setTimeout(100); 

ロールする必要があるSQLクエリを追加します。

 $mclq->addSqlUpdate(" CREATE DATABASE IF NOT EXISTS dbpulse "); $mclq->addSqlUpdate(" CREATE TABLE IF NOT EXISTS dbpulse.normal_summing_sharded ( event_date Date DEFAULT toDate(event_time), event_time DateTime DEFAULT now(), body_id Int32, views Int32 ) ENGINE = ReplicatedSummingMergeTree('/clickhouse/tables/{pulse_replica}/pulse/normal_summing_sharded', '{replica}', event_date, (event_date, event_time, body_id), 8192) "); 

エラーの場合にロールバックを実行するSQLクエリを追加します。

 $mclq->addSqlDowngrade(' DROP TABLE IF EXISTS dbpulse.normal_summing_sharded '); $mclq->addSqlDowngrade(' DROP DATABASE IF EXISTS dbpulse '); 

ローリング移行には2つの戦略があります。

• 次のSQLクエリへの移行とともに、個々のSQLクエリを1つのサーバーに送信します。
• すべてのSQLクエリを1つのサーバーに送信し、次のサーバーに移行します。

エラーが発生した場合、次のオプションが可能です。

• アップグレード要求が既に行われているすべてのノードへのダウングレード要求の実行。
• 他のサーバーにアップグレード要求を送信する前に待機します。
• エラーが発生した場合にすべてのサーバーでダウングレード要求を実行します。

クラスターの状態が不明な場合のエラーは、特別な場所を取ります。

• 接続タイムアウトエラー。
• サーバーとの通信エラー。

移行を実行する際のPHPコードの動作の原則は次のとおりです。

 //   IP-   $node_hosts=$this->getClusterNodes($migration->getClusterName()); //  downgrade- $sql_down=$migration->getSqlDowngrade(); //  upgrade- $sql_up=$migration->getSqlUpdate(); //  upgrade-    ,   ,  downgrade- $need_undo=false; $undo_ip=[]; foreach ($sql_up as $s_u) { foreach ($node_hosts as $node) { //  upgrade- $state=$this->client($node)->write($s_u); if ($state->isError()) { $need_undo = true; } else { // OK } if ($need_undo) { //   ,    $undo_ip[$node]=1; break; } } } //    upgrade-     if (!$need_undo) { return true; // OK } 

エラーが発生した場合、ダウングレード要求がクラスターのすべてのノードに送信されます。

 foreach ($node_hosts as $node) { foreach ($sql_down as $s_u) { try{ $st=$this->client($node)->write($s_u); } catch (Exception $E) { //       downgrade- } } } 

ClickHouseでの経験に関する一連の資料を継続します。

結論として、小さな調査を実施したいと思います。