跳转到主要内容

​
简介

ClickHouse 处理查询的速度极快，但这些查询是如何 在多台服务器之间分发并并行处理的？
在本指南中，我们将先介绍 ClickHouse 如何通过 分布式表将查询分发到多个分片，然后再说明一个查询如何利用 多个副本来执行。

​
分片架构

在 shared-nothing 架构中，集群通常会拆分为 多个分片，每个分片都包含整体数据的一个子集。分布式表 位于这些分片之上，为完整数据提供统一视图。 读取请求既可以发送到本地表，此时查询只会在 指定分片上执行；也可以发送到分布式表，在这种情况下， 每个分片都会执行相应的查询。被查询的分布式表所在服务器 会对数据进行聚合，并向客户端返回响应： 上图展示了客户端查询分布式表时会发生什么：
  1. SELECT 查询会先被发送到某个节点上的分布式表， 这个节点可以是随机选中的 (通过轮询策略) ，也可以由负载均衡器 路由到特定服务器。此时，该节点将充当协调器。
  2. 该节点会根据分布式表中指定的信息，定位每个需要执行查询的分片， 然后将查询发送到各个分片。
  3. 每个分片都会在本地读取、过滤并聚合数据， 然后将可合并状态返回给协调器。
  4. 协调器会合并数据，然后将响应返回给客户端。
加入副本后，整体流程基本相同，唯一 的区别是每个分片中只会有一个副本执行该查询。 这意味着可以并行处理更多查询。

​
非分片架构

ClickHouse Cloud 的架构与上文介绍的架构有很大不同。 (更多详情请参见”ClickHouse Cloud Architecture”。) 借助计算与存储分离，以及几乎无限的 存储容量，对分片的需求已不再那么重要。 下图展示了 ClickHouse Cloud 的架构： 这种架构让我们几乎可以瞬间添加和移除副本， 从而确保集群具备极高的可扩展性。右侧所示的 ClickHouse Keeper 集群为元数据提供了单一事实来源。 副本可以从 ClickHouse Keeper 集群拉取元数据，并保持相同的数据。数据本身存储在 对象存储中，而 SSD 缓存则可以加速查询。 但现在，我们如何将查询执行分布到多个服务器上呢？在 分片架构中，这一点很容易理解，因为每个分片实际上都可以 在部分数据子集上执行查询。那么在没有分片的情况下，它是如何工作的呢？

​
介绍并行副本

要通过多台服务器并行执行查询，首先需要能够将其中一台服务器指定为协调器。协调器负责创建待执行的任务列表，确保这些任务全部得到执行、完成聚合，并将结果返回给客户端。与大多数分布式系统一样，这一角色通常由接收初始查询的节点承担。我们还需要定义工作单元。在分片架构中，工作单元是分片，即数据的一个子集。而在并行副本中，我们将使用表中的一小部分，也就是粒度，作为工作单元。 现在，让我们借助下图看看它在实际中是如何工作的： 使用并行副本时：
  1. 客户端发出的查询先经过负载均衡器，然后被发送到某个节点。该节点将成为此次查询的协调器。
  2. 该节点会分析每个 分片 的索引，并选出需要处理的 分片 和粒度。
  3. 协调器会将工作负载拆分成一组可分配给不同副本的粒度。
  4. 每组粒度会由相应的副本处理，完成后会将可合并状态发送给协调器。
  5. 最后，协调器会合并来自各个副本的结果，然后将响应返回给客户端。
以上步骤概述了并行副本在理论上的工作方式。 然而在实际中，有很多因素可能导致这套逻辑无法完美运行：
  1. 某些副本可能不可用。
  2. ClickHouse 中的复制是异步的，因此某些副本在某一时刻可能并不拥有相同的 分片。
  3. 需要以某种方式处理副本之间的长尾延迟。
  4. 文件系统缓存会因各个副本上的活动不同而有所差异，这意味着随机分配任务可能会因为缓存局部性而导致性能不够理想。
我们将在接下来的章节中说明如何克服这些因素。

​
通知

为了解决上述列表中的 (1) 和 (2)，我们引入了“通知”这一概念。下面通过下图来说明其工作方式：
  1. 来自客户端的查询经过负载均衡器后，会被发送到某个节点。该节点将成为此次查询的协调器。
  2. 协调器会发送请求，以获取集群中所有副本的通知。 对于某个表当前的 分片 集合，不同副本看到的内容可能略有差异。因此，我们需要 收集这些信息，以避免做出错误的调度决策。
  3. 随后，协调器会利用这些通知确定一组 可分配给不同副本的粒度。以这里为例， 我们可以看到，分片 3 中的粒度没有分配给副本 2， 因为该副本在其通知中没有提供这个 分片。 还要注意的是，没有任务分配给副本 3，因为该 副本没有提供通知。
  4. 当每个副本都在各自负责的粒度子集上完成查询处理， 并将可合并状态发回协调器后， 协调器会合并结果，然后将响应发送给客户端。

​
动态协调

为了解决长尾延迟问题，我们引入了动态协调。这意味着， 不会在一次请求中将所有粒度都发送给某个副本，而是每个副本 都可以向协调器请求一个新任务 (一组待处理的粒度) 。 协调器会根据收到的通知，将相应的一组粒度分配给该副本。 假设当前流程已经进行到这样一个阶段：所有副本都已发送 包含所有 分片 的通知。 下图展示了动态协调的工作方式：
  1. 副本会告知协调器自己可以处理 任务，也可以说明自己能够处理多少工作量。
  2. 协调器将任务分配给各个副本。
  1. 副本 1 和 2 很快就完成了各自的任务。它们 会向协调器请求另一个任务。
  2. 协调器将新任务分配给副本 1 和 2。
  1. 现在所有副本都已完成各自任务的处理。它们 会请求更多任务。
  2. 协调器利用这些通知检查还剩下哪些任务 需要处理，但此时已经没有剩余任务了。
  3. 协调器会通知各个副本，所有内容都已处理完成。 接下来，它会合并所有可合并状态，并返回查询结果。

​
管理缓存局部性

最后一个尚待解决的潜在问题是如何处理缓存局部性。如果同一个查询 会执行多次，如何确保同一个任务总是被路由到同一个 副本？在前面的示例中，任务分配如下：
副本 1副本 2副本 3
分片 1g1, g6, g7g2, g4, g5g3
分片 2g1g2, g4, g5g3
分片 3g1, g6g2, g4, g5g3
为了确保相同的任务会分配到相同的副本上，从而受益于缓存， 这里会进行两步处理。首先，计算分片 + 粒度集合 (即一个任务) 的哈希值；然后，在任务分配时 按副本数量取模。 理论上这听起来不错，但在实际中，如果某个副本突发负载升高，或者 网络状况恶化，而某些任务又始终固定由同一个副本 执行，就可能引入尾延迟。如果 max_parallel_replicas 小于 副本数量，则会随机选择副本来执行查询。

​
任务窃取

如果某个副本处理任务的速度比其他副本慢，其他副本会尝试 “窃取”按哈希原本应分配给该副本的任务，以降低 长尾延迟。

​
限制

此功能存在一些已知限制，其中较为主要的限制记录在 本节中。
如果你发现了一个不属于下述限制的问题，并且 怀疑是由并行副本导致的，请在 GitHub 上使用 标签 comp-parallel-replicas 提交 issue。

​
排查并行副本问题

你可以在 system.query_log 表中查看每个查询使用了哪些设置。你还可以查看 system.events 表，了解 服务器 上发生的所有事件；也可以使用 clusterAllReplicas 表函数查看所有副本上的表 (如果你是 Cloud 用户，请使用 default) 。
Query
Response
system.text_log 表还 包含使用并行副本执行查询时的相关信息：
Query
Response
最后，你还可以使用 EXPLAIN PIPELINE。它可以清晰地展示 ClickHouse 将如何执行查询，以及在执行该 查询时会用到哪些资源。以下面的查询为例：
我们来看一下未使用并行副本时的查询管道：
EXPLAIN PIPELINE (without parallel replica)
现在再看启用并行副本的情况：
EXPLAIN PIPELINE (with parallel replica)