Skip to content
Enable reading and writing data across shards in Managed ClickHouse®

Enable reading and writing data across shards in Managed ClickHouse®

If your Managed ClickHouse® service uses multiple shards, the data is replicated only between nodes of the same shard.

When you read from one node, you see the data from the shard of this node. Creating a distributed table on top of your replicated table, while the distributed table itself doesn’t store data allows you to see data from all the shards but also helps spread the data evenly across all the cluster nodes.

Set up a sharded service with a database

  1. Create a Managed ClickHouse® service with multiple shards.

Note

Shards are created automatically when you create a Managed ClickHouse service. The number of shards that your service gets depends on the plan you select for your service. Startup plans run a single node with one shard, and Business plans run three nodes that hold one shard. On Premium plans, the number of shards is the number of nodes divided by three.

Important

Multi-shard Premium plans are not creatable on every account. During early access, plan availability is granted per organization, and on a test account in ch-gva-2 in July 2026 only startup-8, startup-16 and startup-32 were authorized, all of them single-shard. Check the authorized column for your own organization before you follow this page:

exo dbaas type show clickhouse --plans

If the plan you need is not authorized, ask our support to enable it for your organization. See plan availability.

  1. Create database test_db in your new service.

Create a distributed table

  1. Connect to your database.

  2. Create a table with the MergeTree engine as shown for the cash_flows table in the following example:

    CREATE TABLE test_db.cash_flows
    (
        EventDate DateTime,
        SourceAccount UInt64,
        TargetAccount UInt64,
        Amount Float64
    )
    ENGINE = MergeTree()
    PARTITION BY toYYYYMM(EventDate)
    ORDER BY (EventDate, SourceAccount)

Note

With Managed ClickHouse, you can specify ENGINE either as MergeTree or as ReplicatedMergeTree. Both of them create a ReplicatedMergeTree table.

  1. Create distributed table cash_flows_distributed with the distributed engine:

    CREATE TABLE test_db.cash_flows_distributed AS test_db.cash_flows
    ENGINE = Distributed(test_db, test_db, cash_flows, SourceAccount)

Verify your distributed table

Check if the distributed table you created is available and if you can use it to access your data from all the shards.

  1. Run a read query for the number of table rows:

    SELECT count() FROM test_db.cash_flows_distributed

    As a response to this query, you can expect to receive a number of rows from all the shards. This is because when you connect on one node and read from the distributed table, ClickHouse® aggregates the data from all the shards and returns all of it.

  2. Run a write query to insert new data into the distributed table:

    INSERT INTO test_db.cash_flows_distributed (
      EventDate, SourceAccount, TargetAccount, Amount
    )
    VALUES (
      '2022-01-02 03:04:05', 123, 456, 100.0
    )

When you insert data into the distributed table, ClickHouse® decides on which node the data should be stored and write it to the correct node making sure that a similar volume of data is written on all the nodes.

Note

Both queries above also run on a single-shard Startup plan, but there is only one shard to fan out to, so they read from and write to that shard and nothing is spread. A distributed table only becomes useful on a plan that has more than one shard.

Last updated on