TOC
分布式
使用分布式引擎的表本身不存储任何数据,但允许在多个服务器上进行分布式查询处理。
读操作自动并行化。在读取期间,将使用远程服务器上的表索引(如果有的话)。
分布式引擎支持参数:
- 服务器配置文件中的群集名称
- 远程数据库的名称
- 远程表的名称
- (可选)分片键
- (可选)策略名,它将用于存储异步发送的临时文件
参考:
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
「如果这篇文章对你有用,请支持一下哦」
如果这篇文章对你有用,请支持一下哦
使用微信扫描二维码完成支付