Table sharding 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 primary sharding strategies:

• Classic approach, where the distributed table uses all shards in the cluster.
• Group-based approach, where some shards are grouped together.
• 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 configuration examples for all three sharding methods.

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 for this solution includes:

• Managed Service for ClickHouse® cluster fee, which covers the use of 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

Set up the infrastructureSet up the infrastructure

Set up the clickhouse-clientSet up the clickhouse-client

Install and configure the clickhouse-client for database access.

Create tables with dataCreate tables with data

Let’s assume you need to enable sharding for the hits_v1 table. The create table statement text will depend on your chosen sharding strategy.

Replace the <table_structure> placeholder with column descriptions from this ClickHouse® guide.

Once you enable sharding (by any method), you will be able to send SELECT and INSERT queries to the distributed table. These queries will be processed according to the specified configuration.

In the following examples, we use a random number, rand(), as a sharding key.

Classic shardingClassic sharding

In this example, the distributed table built from hits_v1 uses every shard in the chcluster cluster: shard1, shard2, and shard3.

Before operating the distributed table:

1. Connect to the tutorial database.

2. Create the hits_v1 table on every host in the cluster using the MergeTree engine:

   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, you can use the AS tutorial.hits_v1 expression instead of explicitly stating the table structure since both tables, hits_v1_distributed and hits_v1, reside 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

   You can use the {cluster} macro instead of the cluster ID. When running the query, ClickHouse will automatically substitute it with the ID of the cluster where the CREATE TABLE statement is running.

Group-based shardingGroup-based sharding

In this example:

• We use a single shard group named sgroup.
• The distributed table and its underlying table, hits_v1, are in the same sgroup shard group within the cluster.

Before operating the distributed table:

1. Connect to the tutorial database.

2. Create the hits_v1 table on every host in the sgroup shard group, using the MergeTree engine:

   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, you can use the AS tutorial.hits_v1 expression instead of explicitly stating the table structure since both tables, hits_v1_distributed and hits_v1, reside on the same cluster hosts within a single shard.

Advanced group-based shardingAdvanced group-based sharding

In this example:

1. We use the sgroup and sgroup_data shard groups.
2. The distributed table resides in the sgroup shard group.
3. The hits_v1 underlying table resides in the sgroup_data shard group.

Before operating the distributed table:

1. Connect to the tutorial database.

2. Create the hits_v1 table on every host of the sgroup_data shard group, using the ReplicatedMergeTree engine:

   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 will provide fault tolerance to this solution.

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 the hits_v1_distributed and hits_v1 tables reside on different hosts in separate shards.

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
   

   You can get the host names with the list of ClickHouse® hosts in the cluster.

3. Run one or multiple test queries against this table. For example, you can get the table row count:

   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.