标签：副本 Distributed 分片 服务器 数据 clickhouse 分布式

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

一、创建表 

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

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。