Sharding tables in ClickHouse®

Sharding provides a number of benefits when dealing with high query rates and massive datasets. It works by creating a distributed table that routes queries to underlying tables. You can access data in sharded tables both directly or through the distributed table.

There are three approaches to sharding:

• The classic approach, where the distributed table uses all shards in the cluster.
• The group-based approach, where some shards are grouped together.
• The advanced group-based approach, where shards are divided into two groups: one for the distributed table and the other for the underlying tables.

Below are examples of sharding setup for all three approaches.

For more information, see Sharding in Managed Service for ClickHouse®.

To set up sharding:

1. Create tables with data.
2. Test the tables.

If you no longer need the resources you created, delete them.

Required paid resourcesRequired paid resources

The support cost includes:

• Managed Service for ClickHouse® cluster fee: Using computing resources allocated to hosts (including ZooKeeper hosts) and disk space (see Managed Service for ClickHouse® pricing).
• Fee for using public IP addresses if public access is enabled for cluster hosts (see Virtual Private Cloud pricing).

Getting startedGetting started

Prepare the infrastructurePrepare the infrastructure

Set up clickhouse-clientSet up clickhouse-client

Install and configure clickhouse-client to connect to your database.

Create tables with dataCreate tables with data

Let's assume you need to enable sharding for the hits_v1 table. The text of the table creation query depends on the sharding approach you selected.

For the table structure to substitute instead of <table_structure>, see the ClickHouse® documentation.

Once you engage any of the sharding methods, you can send the SELECT and INSERT queries to the distributed table you created, and they will be processed according to the specified configuration.

The sharding key in the examples is a random number rand().

Classic shardingClassic sharding

In this example, the distributed table that will be created based on hits_v1 uses all the shards of the chcluster cluster: shard1, shard2, shard3.

Before operating the distributed table:

1. Connect to the tutorial database.

2. Create a MergeTree table named hits_v1 that will reside on all the cluster's hosts:

   CREATE TABLE tutorial.hits_v1 ON CLUSTER '{cluster}' ( <table_structure> )
   ENGINE = MergeTree()
   PARTITION BY toYYYYMM(EventDate)
   ORDER BY (CounterID, EventDate, intHash32(UserID))
   SAMPLE BY intHash32(UserID)
   SETTINGS index_granularity = 8192
   

To create a distributed table named hits_v1_distributed in the cluster:

1. Connect to the tutorial database.

2. Create a Distributed table:

   CREATE TABLE tutorial.hits_v1_distributed ON CLUSTER '{cluster}' AS tutorial.hits_v1
   ENGINE = Distributed('{cluster}', tutorial, hits_v1, rand())
   

   Here, instead of explicitly specifying the table structure, you can use the AS tutorial.hits_v1 expression because the hits_v1_distributed and hits_v1 tables are on the same cluster hosts.

   When creating a Distributed table, use chcluster as the cluster ID. You can get it with the list of clusters in the folder.

   Tip

   Instead of the cluster ID, you can use the {cluster} macro: when executing the query, the ID of the cluster the CREATE TABLE operation is being executed in will be substituted automatically.

Group-based shardingGroup-based sharding

In this example:

• One shard group is used named sgroup.
• A distributed table and the underlying table named hits_v1 are in the same cluster shard group named sgroup.

Before operating the distributed table:

1. Connect to the tutorial database.

2. Create a MergeTree table named hits_v1 that uses all the cluster's sgroup shard group hosts:

   CREATE TABLE tutorial.hits_v1 ON CLUSTER sgroup ( <table_structure> )
   ENGINE = MergeTree()
   PARTITION BY toYYYYMM(EventDate)
   ORDER BY (CounterID, EventDate, intHash32(UserID))
   SAMPLE BY intHash32(UserID)
   SETTINGS index_granularity = 8192
   

To create a distributed table named tutorial.hits_v1_distributed in the cluster:

1. Connect to the tutorial database.

2. Create a Distributed table:

   CREATE TABLE tutorial.hits_v1_distributed ON CLUSTER sgroup AS tutorial.hits_v1
   ENGINE = Distributed(sgroup, tutorial, hits_v1, rand())
   

   Here, instead of explicitly specifying the table structure, you can use the AS tutorial.hits_v1 expression because the hits_v1_distributed and hits_v1 tables use the same shard and run on the same hosts.

Advanced group-based shardingAdvanced group-based sharding

In this example:

1. Two shard groups are used: sgroup and sgroup_data.
2. The distributed table is in the shard group named sgroup.
3. The hits_v1 underlying table is in the shard group named sgroup_data.

Before operating the distributed table:

1. Connect to the tutorial database.

2. Create a ReplicatedMergeTree table named hits_v1 that uses all the cluster's sgroup_data shard group hosts:

   CREATE TABLE tutorial.hits_v1 ON CLUSTER sgroup_data ( <table_structure> )
   ENGINE = ReplicatedMergeTree('/tables/{shard}/hits_v1', '{replica}')
   PARTITION BY toYYYYMM(EventDate)
   ORDER BY (CounterID, EventDate, intHash32(UserID))
   SAMPLE BY intHash32(UserID)
   SETTINGS index_granularity = 8192
   

   The ReplicatedMergeTree engine is used for fault tolerance.

To create a distributed table named tutorial.hits_v1_distributed in the cluster:

1. Connect to the tutorial database.

2. Create a Distributed table:

   CREATE TABLE tutorial.hits_v1_distributed ON CLUSTER sgroup ( <table_structure> )
   ENGINE = Distributed(sgroup_data, tutorial, hits_v1, rand())
   

   Here you must explicitly specify the table structure because hits_v1_distributed and hits_v1 use different shards and are on different hosts.

Test the tablesTest the tables

To test your new distributed table named tutorial.hits_v1_distributed:

1. Load the hits_v1 test dataset:

   curl https://storage.yandexcloud.net/doc-files/managed-clickhouse/hits_v1.tsv.xz | unxz --threads=`nproc` > hits_v1.tsv
   

2. Populate the table with test data:

   clickhouse-client \
      --host "<FQDN_of_any_host_with_distributed_table>" \
      --secure \
      --port 9440 \
      --user "<username>" \
      --password "<user_password>" \
      --database "tutorial" \
      --query "INSERT INTO tutorial.hits_v1_distributed FORMAT TSV" \
      --max_insert_block_size=100000 < hits_v1.tsv
   

   To find out the host names, request a list of ClickHouse® hosts in the cluster.

3. Run one or more test queries to this table. For example, you can find out the number of rows in it:

   SELECT count() FROM tutorial.hits_v1_distributed
   

   Result:

   8873898
   

Delete the resources you createdDelete the resources you created

Delete the resources you no longer need to avoid paying for them:

ClickHouse® is a registered trademark of ClickHouse, Inc.