Cluster Configuration Overview

Cluster Deployment Workflow

1. Install ClickHouse server across all node
2. Configure cluster parameters in XML files
3. Create local storage tables on each instance
4. Implement distributed table abstraction layer
5. Server Installation Refer to official installation documentation for CentOS 7 setup.
6. Configuration Setup Update the following files:

<yandex>
  <clickhouse_remote_servers>
    <cluster_3shards_1replicas>
      <shard>
        <replica>
          <host>centf8118.sharding1.db</host>
          <port>9000</port>
        </replica>
      </shard>
      <shard>
        <replica>
          <host>centf8119.sharding2.db</host>
          <port>9000</port>
        </replica>
      </shard>
      <shard>
        <replica>
          <host>centf8120.sharding3.db</host>
          <port>9000</port>
        </replica>
      </shard>
    </cluster_3shards_1replicas>
  </clickhouse_remote_servers>
</yandex>

Verify configuration with:

SELECT * FROM system.clusters;

3. Local Table Creation Data Preparation Using the OnTime flight dataset:

# Download data
for year in {1987..2018}; do
  for month in {1..12}; do
    wget https://transtats.bts.gov/PREZIP/On_Time_Reporting_Carrier_On_Time_Performance_1987_present_${year}_${month}.zip
  done
done

-- Create local table
CREATE TABLE ontime_local (
  Year UInt16,
  Quarter UInt8,
  Month UInt8,
  -- (50+ additional columns)
) ENGINE = MergeTree()
PARTITION BY Year
ORDER BY (Carrier, FlightDate);

4. Distributed Table Implementation

CREATE TABLE ontime_all AS ontime_local
ENGINE = Distributed(cluster_3shards_1replicas, datasets, ontime_local, rand());

Insert data:

INSERT INTO ontime_all SELECT * FROM ontime;

Performance Comparison Single Node Query

SELECT 
  Carrier, 
  count() AS c, 
  round(quantileTDigest(0.99)(DepDelay), 2) AS q 
FROM ontime 
GROUP BY Carrier 
ORDER BY q DESC 
LIMIT 5;

First execution: 3.561 sec | Second execution: 1.214 sec

Distributed Query

SELECT 
  Carrier, 
  count() AS c, 
  round(quantileTDigest(0.99)(DepDelay), 2) AS q 
FROM ontime_all 
GROUP BY Carrier 
ORDER BY q DESC 
LIMIT 5;

First execution: 17.254 sec | Second execution: 0.892 sec

High Availability Configuration Relpicated Table Setup

<cluster_1shards_2replicas>
  <shard>
    <internal_replication>true</internal_replication>
    <replica><host>centf8118.sharding1.db</host></replica>
    <replica><host>centf8119.sharding2.db</host></replica>
  </shard>
</cluster_1shards_2replicas>

CREATE TABLE ontime_replica (
  -- same column definitions
) ENGINE = ReplicatedMergeTree(
  '/clickhouse/tables/ontime', 
  'replica{replica_id}', 
  FlightDate, 
  (Year, FlightDate), 
  8192
);

ZooKeeper Integration

<zookeeper-servers>
  <node><host>zk1.example.com</host><port>2181</port></node>
  <node><host>zk2.example.com</host><port>2181</port></node>
</zookeeper-servers>

Sharding + Replication Architecture

<cluster_3shards_2replicas>
  <shard><replica><host>node1</host></replica><replica><host>node2</host></replica></shard>
  <shard><replica><host>node3</host></replica><replica><host>node4</host></replica></shard>
  <shard><replica><host>node5</host></replica><replica><host>node6</host></replica></shard>
</cluster_3shards_2replicas>

Requires 6 distinct ClickHouse instances for complete redundancy.