title_wip

[Edit this page](https://github.com/ClickHouse/ClickHouse/tree/master/docs/en/engines/table-engines/special/distributed.md)

Distributed engine in Cloud

To create a distributed table engine in ClickHouse Cloud, you can use the [`remote` and `remoteSecure`](/docs/sql-reference/table-functions/remote) table functions.
The `Distributed(...)` syntax cannot be used in ClickHouse Cloud.

Tables with Distributed engine do not store any data of their own, but allow distributed query processing on multiple servers.
Reading is automatically parallelized. During a read, the table indexes on remote servers are used if they exist.

## Creating a table[​](#distributed-creating-a-table "Direct link to Creating a table")

```
CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
    name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
    name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
    ...
) ENGINE = Distributed(cluster, database, table[, sharding_key[, policy_name]])
[SETTINGS name=value, ...]
```

### From a table[​](#distributed-from-a-table "Direct link to From a table")

When the `Distributed` table is pointing to a table on the current server you can adopt that table's schema:

```
CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster] AS [db2.]name2 ENGINE = Distributed(cluster, database, table[, sharding_key[, policy_name]]) [SETTINGS name=value, ...]
```

### Distributed parameters[​](#distributed-parameters "Direct link to Distributed parameters")

|  |  |  |  |  |  |  |  |  |  |  |  |
| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |
| Parameter Description|  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `cluster` The cluster name in the server's config file|  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | | `database` The name of a remote database|  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | | `table` The name of a remote table|  |  |  |  | | --- | --- | --- | --- | | `sharding_key` (Optional) The sharding key.   Specifying the `sharding_key` is necessary for the following:  - For `INSERTs` into a distributed table (as the table engine needs the `sharding_key` to determine how to split the data). However, if `insert_distributed_one_random_shard` setting is enabled, then `INSERTs` do not need the sharding key.- For use with `optimize_skip_unused_shards` as the `sharding_key` is necessary to determine what shards should be queried  | `policy_name` (Optional) The policy name, it will be used to store temporary files for background send | | | | | | | | | | | |

**See Also**

- [distributed\_foreground\_insert](/docs/operations/settings/settings#distributed_foreground_insert) setting
- [MergeTree](/docs/engines/table-engines/mergetree-family/mergetree#table_engine-mergetree-multiple-volumes) for the examples

### Distributed settings[​](#distributed-settings "Direct link to Distributed settings")

|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |
| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |
| Setting Description Default value|  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  |  | | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | | `fsync_after_insert` Do the `fsync` for the file data after background insert to Distributed. Guarantees that the OS flushed the whole inserted data to a file **on the initiator node** disk. `false`| `fsync_directories` Do the `fsync` for directories. Guarantees that the OS refreshed directory metadata after operations related to background inserts on Distributed table (after insert, after sending the data to shard, etc.). `false`| `skip_unavailable_shards` If true, ClickHouse silently skips unavailable shards. Shard is marked as unavailable when: 1) The shard cannot be reached due to a connection failure. 2) Shard is unresolvable through DNS. 3) Table does not exist on the shard. `false`| `bytes_to_throw_insert` If more than this number of compressed bytes will be pending for background `INSERT`, an exception will be thrown. `0` - do not throw. `0`| `bytes_to_delay_insert` If more than this number of compressed bytes will be pending for background INSERT, the query will be delayed. 0 - do not delay. `0`| `max_delay_to_insert` Max delay of inserting data into Distributed table in seconds, if there are a lot of pending bytes for background send. `60`| `background_insert_batch` The same as [`distributed_background_insert_batch`](/docs/operations/settings/settings#distributed_background_insert_batch) `0`| `background_insert_split_batch_on_failure` The same as [`distributed_background_insert_split_batch_on_failure`](/docs/operations/settings/settings#distributed_background_insert_split_batch_on_failure) `0`| `background_insert_sleep_time_ms` The same as [`distributed_background_insert_sleep_time_ms`](/docs/operations/settings/settings#distributed_background_insert_sleep_time_ms) `0`| `background_insert_max_sleep_time_ms` The same as [`distributed_background_insert_max_sleep_time_ms`](/docs/operations/settings/settings#distributed_background_insert_max_sleep_time_ms) `0`| `flush_on_detach` Flush data to remote nodes on `DETACH`/`DROP`/server shutdown. `true` | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | | |

Note

**Durability settings** (`fsync_...`):

- Affect only background `INSERT`s (i.e. `distributed_foreground_insert=false`) when data is first stored on the initiator node disk and later, in the background, when sent to shards.
- May significantly decrease `INSERT` performance
- Affect writing the data stored inside the distributed table folder into the **node which accepted your insert**. If you need to have guarantees of writing data to the underlying MergeTree tables, see durability settings (`...fsync...`) in `system.merge_tree_settings`

For **Insert limit settings** (`..._insert`) see also:

- [`distributed_foreground_insert`](/docs/operations/settings/settings#distributed_foreground_insert) setting
- [`prefer_localhost_replica`](/docs/operations/settings/settings#prefer_localhost_replica) setting
- `bytes_to_throw_insert` handled before `bytes_to_delay_insert`, so you should not set it to the value less then `bytes_to_delay_insert`

**Example**

```
CREATE TABLE hits_all AS hits
ENGINE = Distributed(logs, default, hits[, sharding_key[, policy_name]])
SETTINGS
    fsync_after_insert=0,
    fsync_directories=0;
```

Data will be read from all servers in the `logs` cluster, from the `default.hits` table located on every server in the cluster. Data is not only read but is partially processed on the remote servers (to the extent that this is possible). For example, for a query with `GROUP BY`, data will be aggregated on remote servers, and the intermediate states of aggregate functions will be sent to the requestor server. Then data will be further aggregated.

Instead of the database name, you can use a constant expression that returns a string. For example: `currentDatabase()`.

## Clusters[​](#distributed-clusters "Direct link to Clusters")

Clusters are configured in the [server configuration file](/docs/operations/configuration-files):

```
<remote_servers>
    <logs>
        <!-- Inter-server per-cluster secret for Distributed queries
             default: no secret (no authentication will be performed)

             If set, then Distributed queries will be validated on shards, so at least:
             - such cluster should exist on the shard,
             - such cluster should have the same secret.