分布

    将会从位于«logs»集群中 default.hits 表所有服务器上读取数据。
    远程服务器不仅用于读取数据,还会对尽可能数据做部分处理。
    例如,对于使用 GROUP BY 的查询,数据首先在远程服务器聚合,之后返回聚合函数的中间状态给查询请求的服务器。再在请求的服务器上进一步汇总数据。

    数据库名参数除了用数据库名之外,也可用返回字符串的常量表达式。例如:currentDatabase()。

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

    集群示例配置如下:

    1. <logs>
    2. <shard>
    3. <!-- Optional. Shard weight when writing data. Default: 1. -->
    4. <weight>1</weight>
    5. <!-- Optional. Whether to write data to just one of the replicas. Default: false (write data to all replicas). -->
    6. <internal_replication>false</internal_replication>
    7. <replica>
    8. <host>example01-01-1</host>
    9. <port>9000</port>
    10. <replica>
    11. <host>example01-01-2</host>
    12. <port>9000</port>
    13. </replica>
    14. <shard>
    15. <weight>2</weight>
    16. <internal_replication>false</internal_replication>
    17. <replica>
    18. <host>example01-02-1</host>
    19. </replica>
    20. <replica>
    21. <host>example01-02-2</host>
    22. <secure>1</secure>
    23. <port>9440</port>
    24. </replica>
    25. </shard>
    26. </logs>

    这里定义了一个名为’logs’的集群,它由两个分片组成,每个分片包含两个副本。
    分片是指包含数据不同部分的服务器(要读取所有数据,必须访问所有分片)。
    副本是存储复制数据的服务器(要读取所有数据,访问任一副本上的数据即可)。

    集群名称不能包含点号。

    每个服务器需要指定 hostport,和可选的 userpasswordsecurecompression 的参数:
    - host – 远程服务器地址。可以域名、IPv4或IPv6。如果指定域名,则服务在启动时发起一个 DNS 请求,并且请求结果会在服务器运行期间一直被记录。如果 DNS 请求失败,则服务不会启动。如果你修改了 DNS 记录,则需要重启服务。
    - port – 消息传递的 TCP 端口(「tcp_port」配置通常设为 9000)。不要跟 http_port 混淆。
    - user – 用于连接远程服务器的用户名。默认值:default。该用户必须有权限访问该远程服务器。访问权限配置在 users.xml 文件中。更多信息,请查看«访问权限»部分。
    - password – 用于连接远程服务器的密码。默认值:空字符串。
    - secure – 是否使用ssl进行连接,设为true时,通常也应该设置 port = 9440。服务器也要监听 <tcp_port_secure>9440</tcp_port_secure> 并有正确的证书。
    - compression - 是否使用数据压缩。默认值:true。

    配置了副本,读取操作会从每个分片里选择一个可用的副本。可配置负载平衡算法(挑选副本的方式) - 请参阅«load_balancing»设置。
    如果跟服务器的连接不可用,则在尝试短超时的重连。如果重连失败,则选择下一个副本,依此类推。如果跟所有副本的连接尝试都失败,则尝试用相同的方式再重复几次。
    该机制有利于系统可用性,但不保证完全容错:如有远程服务器能够接受连接,但无法正常工作或状况不佳。

    你可以配置一个(这种情况下,查询操作更应该称为远程查询,而不是分布式查询)或任意多个分片。在每个分片中,可以配置一个或任意多个副本。不同分片可配置不同数量的副本。

    要查看集群,可使用«system.clusters»表。

    通过分布式引擎可以像使用本地服务器一样使用集群。但是,集群不是自动扩展的:你必须编写集群配置到服务器配置文件中(最好,给所有集群的服务器写上完整配置)。

    不支持用分布式表查询别的分布式表(除非该表只有一个分片)。或者说,要用分布表查查询«最终»的数据表。

    分布式引擎需要将集群信息写入配置文件。配置文件中的集群信息会即时更新,无需重启服务器。如果你每次是要向不确定的一组分片和副本发送查询,则不适合创建分布式表 - 而应该使用«远程»表函数。 请参阅«表函数»部分。

    向集群写数据的方法有两种:

    一,自已指定要将哪些数据写入哪些服务器,并直接在每个分片上执行写入。换句话说,在分布式表上«查询»,在数据表上 INSERT。
    这是最灵活的解决方案 – 你可以使用任何分片方案,对于复杂业务特性的需求,这可能是非常重要的。
    这也是最佳解决方案,因为数据可以完全独立地写入不同的分片。

    二,在分布式表上执行 INSERT。在这种情况下,分布式表会跨服务器分发插入数据。
    为了写入分布式表,必须要配置分片键(最后一个参数)。当然,如果只有一个分片,则写操作在没有分片键的情况下也能工作,因为这种情况下分片键没有意义。

    每个分片都可以在配置文件中定义权重。默认情况下,权重等于1。数据依据分片权重按比例分发到分片上。例如,如果有两个分片,第一个分片的权重是9,而第二个分片的权重是10,则发送 9 / 19 的行到第一个分片, 10 / 19 的行到第二个分片。

    分片可在配置文件中定义 ‘internal_replication’ 参数。

    若此参数设置为«false»(默认值),写操作会将数据写入所有副本。实质上,这意味着要分布式表本身来复制数据。这种方式不如使用复制表的好,因为不会检查副本的一致性,并且随着时间的推移,副本数据可能会有些不一样。

    选择将一行数据发送到哪个分片的方法是,首先计算分片表达式,然后将这个计算结果除以所有分片的权重总和得到余数。该行会发送到那个包含该余数的从’prev_weight’到’prev_weights + weight’的半闭半开区间对应的分片上,其中 ‘prev_weights’ 是该分片前面的所有分片的权重和,‘weight’ 是该分片的权重。例如,如果有两个分片,第一个分片权重为9,而第二个分片权重为10,则余数在 [0,9) 中的行发给第一个分片,余数在 [9,19) 中的行发给第二个分片。

    分片表达式可以是由常量和表列组成的任何返回整数表达式。例如,您可以使用表达式 ‘rand()’ 来随机分配数据,或者使用 ‘UserID’ 来按用户 ID 的余数分布(相同用户的数据将分配到单个分片上,这可降低带有用户信息的 IN 和 JOIN 的语句运行的复杂度)。如果该列数据分布不够均匀,可以将其包装在散列函数中:intHash64(UserID)。

    这种简单的用余数来选择分片的方案是有局限的,并不总适用。它适用于中型和大型数据(数十台服务器)的场景,但不适用于巨量数据(数百台或更多服务器)的场景。后一种情况下,应根据业务特性需求考虑的分片方案,而不是直接用分布式表的多分片。

    SELECT 查询会被发送到所有分片,并且无论数据在分片中如何分布(即使数据完全随机分布)都可正常工作。添加新分片时,不必将旧数据传输到该分片。你可以给新分片分配大权重然后写新数据 - 数据可能会稍分布不均,但查询会正确高效地运行。

    下面的情况,你需要关注分片方案:

    • 使用需要特定键连接数据( IN 或 JOIN )的查询。如果数据是用该键进行分片,则应使用本地 IN 或 JOIN 而不是 GLOBAL IN 或 GLOBAL JOIN,这样效率更高。

    数据是异步写入的。对于分布式表的 INSERT,数据块只写本地文件系统。之后会尽快地在后台发送到远程服务器。你可以通过查看表目录中的文件列表(等待发送的数据)来检查数据是否成功发送:/var/lib/clickhouse/data/database/table/ 。

    如果在 INSERT 到分布式表时服务器节点丢失或重启(如,设备故障),则插入的数据可能会丢失。如果在表目录中检测到损坏的数据分片,则会将其转移到«broken»子目录,并不再使用。

    启用 max_parallel_replicas 选项后,会在分表的所有副本上并行查询处理。更多信息,请参阅«设置,max_parallel_replicas»部分。