Table sharding in ClickHouse®

In a sharded ClickHouse® cluster, you can distribute table data across all cluster shards or only some of them, i.e., a shard group.

You can select one of the following three topologies to implement distributed storage of table data on cluster shards:

• All shards: the local and distributed tables are created across all shards within a cluster.
• Single shard group: the local and distributed tables are created on all shards within a group.
• Two shard groups: the local table is created on the shards of one group, while the distributed table is deployed on shards of the second group.

To split table data into shards:

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

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

Getting startedGetting started

Set up your infrastructureSet up your 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 we need to distribute the hits_v1 table data across shards.

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

In the following examples, we use the rand() expression as a sharding key to randomly distribute data across shards.

All shardsAll shards

In this example, the distributed and local tables are created on all cluster shards: shard1, shard2, and shard3.

Before creating a distributed table:

1. Connect to the tutorial database.

2. Create the hits_v1 MergeTree table on all cluster 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, 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.

Single shard groupSingle shard group

In this example, the distributed and local tables are created within a single shard group, sgroup.

Before creating a distributed table:

1. Connect to the tutorial database.

2. Create the hits_v1 MergeTree table on all hosts within the sgroup shard group:

   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.

Two shard groupsTwo shard groups

In this example:

• The distributed table is created within the sgroup shard group.
• The local table is created within the sgroup_data shard group.

Before creating a distributed table:

1. Connect to the tutorial database.

2. Create the hits_v1 ReplicatedMergeTree local table on all hosts within the sgroup_data shard group:

   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 the Distributed table on all hosts within the sgroup shard group:

   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.