首页 > 其他分享> > clickhouse的分布式Distributed表引擎

clickhouse的分布式Distributed表引擎

2022-04-06 15:01:50 作者：互联网

　　具有分布式引擎的表不存储自己的任何数据，但允许在多个服务器上进行分布式查询处理。读取是自动并行的。在读取期间，将使用远程服务器上的表索引（如果有的话）。

一、创建表

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, ...]

　　1.来源表

　　当Distributed表指向当前服务器上的表时，可以采用该表的模式：

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, ...]

　　分布式参数

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

　　分布式设置

fsync_after_insert-fsync异步插入分布式后的文件数据。保证操作系统将整个插入的数据刷新到启动器节点磁盘上的文件中。
fsync_directories-fsync为目录做。保证操作系统在分布式表上与异步插入相关的操作后（插入后、将数据发送到分片后等）刷新目录元数据。
bytes_to_throw_insert- 如果超过此数量的压缩字节将等待异步 INSERT，则会引发异常。0 - 不扔。默认为 0。
bytes_to_delay_insert- 如果超过此数量的压缩字节将等待异步 INSERT，则查询将被延迟。0 - 不延迟。默认为 0。
max_delay_to_insert- 如果有大量用于异步发送的待处理字节，则以秒为单位将数据插入分布式表的最大延迟。默认 60。
monitor_batch_inserts- 与Distributed_directory_monitor_batch_inserts相同
monitor_split_batch_on_failure- 与distributed_directory_monitor_split_batch_on_failure相同
monitor_sleep_time_ms- 与Distributed_directory_monitor_sleep_time_ms相同
monitor_max_sleep_time_ms- 与Distributed_directory_monitor_max_sleep_time_ms相同

　　 注意：

　　- 当数据首先存储在启动器节点磁盘上，然后异步发送到分片时，仅影响异步 INSERT（即 `insert_distributed_sync=false`）。
　　- 可能会显着降低刀片的性能
　　- 影响将分布式表文件夹中存储的数据写入接受插入的**节点**。如果需要保证将数据写入底层 MergeTree 表

例子

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

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

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

二、集群

　　集群在服务器配置文件中配置：

<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.

             And also (and which is more important), the initial_user will
             be used as current user for the query.
        -->
        <!-- <secret></secret> -->
        <shard>
            <!-- Optional. Shard weight when writing data. Default: 1. -->
            <weight>1</weight>
            <!-- Optional. Whether to write data to just one of the replicas. Default: false (write data to all replicas). -->
            <internal_replication>false</internal_replication>
            <replica>
                <!-- Optional. Priority of the replica for load balancing (see also load_balancing setting). Default: 1 (less value has more priority). -->
                <priority>1</priority>
                <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/TLS 连接。通常还需要指定端口（默认安全端口是9440）。服务器应该监听<tcp_port_secure>9440</tcp_port_secure>并配置正确的证书。
compression- 使用数据压缩。默认值：true。

　　指定副本时，读取时将为每个分片选择一个可用副本。可以配置负载平衡算法（访问哪个副本的首选项）。如果未建立与服务器的连接，则将尝试连接一个短暂的超时。如果连接失败，将选择下一个副本，以此类推所有副本。如果所有副本的连接尝试都失败，则尝试以相同的方式重复几次。这有利于弹性，但不提供完整的容错能力：远程服务器可能接受连接，但可能无法正常工作，或者工作不佳。

　　可以仅指定一个分片（在这种情况下，查询处理应称为远程，而不是分布式）或最多指定任意数量的分片。在每个分片中，可以指定从一个到任意数量的副本。可以为每个分片指定不同数量的副本。

　　可以在配置中指定任意数量的集群。

　　要查看的集群，请使用该system.clusters表。

　　该Distributed引擎允许使用像本地服务器这样的集群。但是，集群的配置不能动态指定，必须在服务器配置文件中进行配置。通常，集群中的所有服务器都将具有相同的集群配置（尽管这不是必需的）。配置文件中的集群会即时更新，无需重新启动服务器。

　　如果每次都需要向一组未知的分片和副本发送查询，则无需创建Distributed表 -remote而是使用 table 函数。

三、写入数据

　　1.将数据写入集群有两种方法：

　　1）可以定义将哪些数据写入哪些服务器并直接在每个分片上执行写入。换句话说，对表所指向INSERT的集群中的远程表执行直接语句。Distributed这是最灵活的解决方案，因为可以使用任何分片方案，即使是由于主题领域的要求而并非微不足道的分片方案。这也是最优化的解决方案，因为数据可以完全独立地写入不同的分片。

　　2）可以在表上执行INSERT语句Distributed。在这种情况下，表将在服务器本身之间分配插入的数据。为了写入Distributed表，它必须sharding_key配置参数（除非只有一个分片）。

2.注意：

　　每个分片都可以<weight>在配置文件中定义。默认情况下，权重为1。数据以与分片权重成比例的数量分布在分片上。将所有分片权重相加，然后将每个分片的权重除以总和，以确定每个分片的比例。例如，如果有两个分片，第一个的权重为 1，而第二个的权重为 2，第一个将被发送三分之一 (1 / 3) 的插入行，第二个将被发送三分之二 (2 / 3）。

　　每个分片都可以internal_replication在配置文件中定义参数。如果此参数设置为true，则写入操作会选择第一个健康的副本并向其写入数据。如果表基础的Distributed表是复制表（例如任何Replicated*MergeTree表引擎），请使用此选项。其中一个表副本将接收写入，并将自动复制到其他副本。

　　如果internal_replication设置为false（默认值），则将数据写入所有副本。在这种情况下，Distributed表本身会复制数据。这比使用复制表更糟糕，因为不检查副本的一致性，并且随着时间的推移，它们将包含稍微不同的数据。

　　为了选择将一行数据发送到的分片，分析分片表达式，并将其除以分片的总权重得到余数。prev_weights该行被发送到对应于余数从到的半区间的分片prev_weights + weight，其中prev_weights是编号最小的分片的总权重，并且weight是该分片的权重。例如，如果有两个分片，第一个的权重为 9，而第二个的权重为 10，则该行将发送到第一个分片以获取范围 [0, 9) 中的余数，并发送到第二个用于范围 [9, 19) 的余数。

　　分片表达式可以是返回整数的常量和表列中的任何表达式。例如，可以使用表达式rand()进行数据的随机分布，或者UserID通过除以用户 ID 的余数进行分布（然后单个用户的数据将驻留在单个分片上，这简化了运行IN和JOIN按用户）。如果其中一列分布不够均匀，可以将其包装在哈希函数中，例如intHash64(UserID).

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

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

使用需要通过特定键连接数据（IN或）的查询。JOIN如果数据被这个键分片，你可以使用 local INorJOIN代替GLOBAL INor GLOBAL JOIN，这样效率更高。
使用大量服务器（数百台或更多）和大量小型查询，例如，查询单个客户（例如网站、广告商或合作伙伴）的数据。为了使小查询不影响整个集群，将单个客户端的数据定位在单个分片上是有意义的。或者，可以设置双层分片：将整个集群划分为“层”，其中一层可能由多个分片组成。单个客户端的数据位于单个层上，但可以根据需要将分片添加到层中，并且数据在其中随机分布。Distributed为每一层创建表，并为全局查询创建一个共享分布式表。

　　数据是异步写入的。当插入表中时，数据块只是写入本地文件系统。数据会尽快在后台发送到远程服务器。发送数据的周期由distributed_directory_monitor_sleep_time_ms和distributed_directory_monitor_max_sleep_time_ms设置管理。Distributed引擎会单独发送每个包含插入数据的文件，但可以使用 Distributed_directory_monitor_batch_inserts 设置启用批量发送文件。此设置通过更好地利用本地服务器和网络资源来提高集群性能。应该通过查看表目录中的文件列表（等待发送的数据）来检查数据是否发送成功：/var/lib/clickhouse/data/database/table/. 执行后台任务的线程数可以通过background_distributed_schedule_pool_size设置来设置。

　　INSERT如果服务器停止存在或在访问表后粗略重新启动（例如，由于硬件故障）Distributed，则插入的数据可能会丢失。如果在表目录中检测到损坏的数据部分，则将其转移到broken子目录中，不再使用。

　　3.应用

　　直接写分布式表的优点自然是可以让ClickHouse控制数据到分片的路由，而缺点：

数据是先写到一个分布式表的实例中并缓存起来，再逐渐分发到各个分片上去，实际是双写了数据（写入放大），浪费资源；
数据写入默认是异步的，短时间内可能造成不一致；
目标表中会产生较多的小parts，使merge（即compaction）过程压力增大。

　　相对而言，直接写本地表是同步操作，更快，parts的大小也比较合适，但是就要求应用层额外实现sharding和路由逻辑，如轮询或者随机等。

　　以下为分布式表插入流程图：

四、读取数据

　　查询Distributed表时，SELECT查询被发送到所有分片并且无论数据如何在分片中分布（它们可以完全随机分布）都可以工作。添加新分片时，不必将旧数据传输到其中。相反，可以通过使用更重的权重向其写入新数据——数据将稍微不均匀地分布，但查询将正确有效地工作。

　　启用该max_parallel_replicas选项后，查询处理将在单个分片内的所有副本中并行处理。

　　在分布式表上执行查询的流程简图如下所示。发出查询后，各个实例之间会交换自己持有的分片的表数据，最终汇总到同一个实例上返回给用户。

五、虚拟列

　　_shard_num— 包含shard_num表中的值system.clusters。类型：UInt32。

　　注意

　　由于远程和集群表功能在内部创建临时分布式表，_shard_num因此在那里也可用。

六、总结

　　ClickHouse分布式表的本质并不是一张表，而是一些本地物理表（分片）的分布式视图，本身并不存储数据。

　　在生产环境中总是推荐写本地表、读分布式表。

标签：副本,Distributed,分片,服务器,数据,clickhouse,分布式
来源： https://www.cnblogs.com/MrYang-11-GetKnow/p/16106389.html