Sharding tables in ClickHouse®

Sharding provides a number of benefits for coping with a high query rate and large data sets. It works by creating a distributed table that routes queries to underlying tables. You can access data in sharded tables both directly and through the distributed table.

There are three approaches to sharding:

• Classic approach, when the distributed table uses all shards in the cluster.
• Regular group-based approach, when some shards are combined into a group.
• Advanced group-based approach, when shards are split into two groups: one group is created for the distributed table and another group is created for underlying tables.

Below are examples of sharding setup for each of the 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.

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 that you selected.

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

Once you enable sharding by any of the 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 a 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 hosts in the cluster.

   When creating a Distributed table, use chcluster as the cluster ID. You can get it with a 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.

Sharding using shard groupsSharding using shard groups

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 a 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 namedtutorial.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 sharding using shard groupsAdvanced sharding using shard groups

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 a 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 ensures fault tolerance.

To create a distributed table namedtutorial.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.