分布式

使用分布式引擎的表本身不存储任何数据，但允许在多个服务器上进行分布式查询处理。

读操作自动并行化。在读取期间，将使用远程服务器上的表索引（如果有的话）。

分布式引擎支持参数：

服务器配置文件中的群集名称
远程数据库的名称
远程表的名称
（可选）分片键
（可选）策略名，它将用于存储异步发送的临时文件

参考：

insert_distributed_sync设置
MergeTree的例子

例子：

Distributed(logs, default, hits[, sharding_key[, policy_name]])

数据将从logs集群中的所有服务器读取，从位于集群中每个服务器上的default.hits表读取。数据不仅被读取，而且在远程服务器进行部分处理（在一定程度上这是可能的）。例如，对于具有GROUP BY的查询，将在远程服务器上聚合数据，并将聚合函数的中间状态发送到发起请求服务器。然后进一步聚合数据。

可以使用返回字符串的常量表达式来代替数据库名。例如：currentDatabase()。

logs — 服务器配置文件中的集群名称。

集群像这样设置：

<remote_servers>
    <logs>
        <shard>
            <!-- 可选的。写入数据时分片的权重。默认值：1。 -->
            <weight>1</weight>
            <!-- 可选的。是否只将数据写入其中一个副本。默认：false（将数据写入所有副本）。 -->
            <internal_replication>false</internal_replication>
            <replica>
                <host>example01-01-1</host>
                <port>9000</port>
            </replica>
            <replica>
                <host>example01-01-2</host>
                <port>9000</port>
            </replica>
        </shard>
        <shard>
            <weight>2</weight>
            <internal_replication>false</internal_replication>
            <replica>
                <host>example01-02-1</host>
                <port>9000</port>
            </replica>
            <replica>
                <host>example01-02-2</host>
                <secure>1</secure>
                <port>9440</port>
            </replica>
        </shard>
    </logs>
</remote_servers>

在里，定义了名称为“logs”的集群，该集群由两个分片组成，每个分片包含两个副本。分片是指包含数据的不同部分的服务器（为了读取所有数据，必须访问所有分片）。副本是指副本服务器（为了读取所有数据，您可以访问任何一个副本上的数据）。

集群名称不能包含点。

必填参数host，port，以及可选参数user，password，secure，compression 为每个服务器指定：

host – 远程服务器的地址。可以使用域名、IPv4或IPv6地址。如果指定了域，服务器在启动时发出DNS请求，并且只要服务器在运行，结果就会被存储。如果DNS请求失败，服务器将无法启动。如果您更改了DNS记录，请重新启动服务器。
port – 互相通信的TCP端口（配置中的'tcp_port'，通常设置为9000）。不要将它与http_port混淆。
user – 连接到远程服务器的用户名称。默认值：default。此用户必须具有连接到指定服务器的访问权限。访问在users.xml文件中配置。有关更多信息，请参考访问权限一节。
password – 连接到远程服务器的密码（非掩码）。默认值：空字符串。
secure – 使用ssl连接，通常还应该定义port = 9440。服务器应该监听<tcp_port_secure>9440</tcp_port_secure>并拥有正确的证书。
compression – 使用数据压缩。默认值：true。

在使用复制时，读取每个分片时会选择一个可用副本。可以配置负载均衡算法（首选访问哪个副本）— 请参考load_balanced设置。如果与服务器的连接没有建立，将使用短超时尝试连接。如果连接失败，则将选择下一个副本，以此类推。如果所有副本的连接尝试都失败，则尝试将以相同的方式重复几次。这有利于弹性，但不提供完全的容错：远程服务器可能接受连接，但可能无法工作，或者工作得很差。

您可以只指定一个分片（在本例中，查询处理应该称为远程，而不是分布式），或者指定任意数量的分片。在每个shard中，您可以指定从一个到任意数量的副本。您可以为每个分片指定不同数量的副本。

您可以在配置中指定多个的集群。

要查看你的集群，请使用‘system.clusters’表。

分布式引擎使得集群像本地服务器一样工作。但是，集群是不可扩展的：您必须在服务器配置文件中写入它的配置（对于所有集群服务器来说，甚至更好）。

分布式引擎需要将集群写入配置文件。配置文件中的集群将动态更新，而无需重新启动服务器。如果每次都需要向一组未知的分片和副本集合发送查询，不需要创建分布式表 — 使用remote表函数替代。参考Table functions章节。

向集群写入数据有两种方法：

首先，明确要向哪些服务器写入数据，并直接在每个分片上执行写入操作。换句话说，在分布式表查看要写入的表，执行INSERT。这是最灵活的解决方案，因为可以使用任何分片方案，由于主体范围的需求，分片方案可能非常重要。这也是最优的解决方案，因为数据可以完全独立地写入不同的分片。

其次，可以在分布式表中执行INSERT。在这种情况下，表将把插入的数据分布到服务器本身。为了写入一个分布式表，它必须有一个分片键集合（最后一个参数）。此外，如果只有一个分片，那么写操作就可以在不指定分片键的情况下工作，因为在这种情况下分片键没有任何意义。

每个分片可以在配置文件中定义一个权重。默认情况下，取重等于1。数据在各个分片之间按分片权重成比例的分布。例如，如果有两个分片，第一个分片的权重为9，第二个分片的权重为10，那么将给第一个分片发送9 / 19行，将给第二个分片发送10 / 19行。

每个分片都可以在配置文件中定义“internal_replication”参数。

如果该参数设置为“true”，则写入操作将选择第一个健康的副本并向其写入数据。如果分布式表“查看”复制表，请使用此方法替代。换句话说，如果要写入数据的表要复制它们本身。

如果将其设置为“false”(默认值)，数据将被写入所有副本。实际上，这意味着分布式表自身复制数据。这比使用复制表更糟糕，因为没有检查副本的一致性，随着时间的推移，它们将包含略有不同的数据。

选择分片然后向分片发送一行数据，分析分片表达式，余数来自分片权重除所有分片的总权重。

这行数据被发送到从'prev_weight'到'prev_weights + weight '的取余的半区间相对应的分片，其中'prev_weights'是带有最小数字的分片的总权重，而'weight'是这个分片的权重。

例如，如果有两个分片，第一分片权重是9而第二个分片权重是10，取余在[0, 9)范围内这样数据被发往第一个分片，取余在[9, 19)范围发往第二个分片。

分片表达式可以是来自常数和返回整数的表中的列的任何表达式。例如，您可以使用表达式“rand()”来表示数据的随机分布，或者通过用户ID的余数来使用“UserID”分派（然后单个用户的数据将存放在一个分片上，这简化了通过用户运行IN和JOIN）。如果其中一列分布得不够均匀，可以将其放在一个散列函数中：intHash64(UserID)。

一个简单提示：除法是有限制的解决方案，它并不总是合适的。它适用于中型和大型数据量（数十个服务器），但不适用于非常大的数据量（数百个或更多服务器）。在后一种情况下，使用主题领域所需的分片方案，而不是在分布式表中使用条目。

SELECT查询被发送到所有的分片并工作，而不管数据是如何分布在各个分片上的（它们可以完全随机分布）。添加新新的分片时，不必给它传输旧数据。您可以用更大的权重写入新数据 —— 数据将稍微不均匀分布，但是查询将正确且有效地工作。

在下列情况下，你应该关注分片方案：

查询需要通过指定的key关联数据（IN或JOIN）。如果数据是由这个key分片的，那么可以使用本地IN或JOIN来替代全局IN或全局JOIN，这样效率更高。
使用大量的服务器（数百个或更多）进行大量的小查询（个人客户端的查询——网站、广告商或合作伙伴）。为了让小查询不影响整个集群，在单个分片上存储单个客户端的数据是有意义的。或者，就像我们在Yandex.Metrica上所做的那样。您可以设置双层切分：将整个集群划分为“层”，其中一个层可能由多个分片组成。单个客户端的数据位于单个层上，但是可以根据需要将分片添加到层中，并且数据在其中随机分布。为每个层创建分布式表，并为全局查询创建一个共享分布式表。

数据是异步写入的。当插入到表中时，数据块被写入本地文件系统。数据在后台被尽快地发送到远程服务器。发送数据的周期由distributed_directory_monitor_sleep_time_ms和distributed_directory_monitor_max_sleep_time_ms设置管理。分布式引擎分别发送每个插入的数据文件，但是您可以通过distributed_directory_monitor_batch_insert设置启用批处理发送文件。该设置通过更好地利用本地服务器和网络资源来提高集群性能。您应该通过检查表目录中的文件列表（等待发送的数据）来检查数据是否发送成功：/var/lib/clickhouse/data/database/table/。

如果在向分布式表插入数据后服务器不复存在或发生了粗略的重启（例如，在设备故障之后），插入的数据可能会丢失。如果在表目录中检测到损坏的数据片段，它将被转移到“broken”子目录中，不再使用。

当启用max_parallel_replicas选项时，查询在单个分片内的所有副本上并行处理。有关更多信息，请参考max_parallel_replicas章节。

虚拟列

_shard_num — 包含shard_num（来自system.clusters）。类型：UInt32

注意：由于remote/cluster表函数在内部创建一个同分布式引擎相同的临时实例，所以这里也可以使用_shard_num。

参考

Virtual columns

「如果这篇文章对你有用，请支持一下哦」

ClickHouse官方文档翻译_引擎-表引擎-特殊-分布式表

TOC

分布式

虚拟列

FEATURED TAGS

FRIENDS