ClickHouse Keeper: 一个用 C++ 编写的 ZooKeeper 替代品

介绍

ClickHouse是用于实时应用和分析的最快且资源利用率最高的开源数据库。ClickHouseKeeper是ClickHouse的一个组件，是ZooKeeper的快速、更节省资源和功能丰富的替代品。这个开源组件提供了一个高度可靠的元数据存储，以及协调和同步机制。最初是为在自建集群或托管的ClickHouse系统中使用而开发的。然而，我们相信其他社区也能在他们的项目中用例中从中受益。

在这篇文章中，我们描述了ClickHouseKeeper的动机、优势和开发，并预览了我们计划的下一步改进。此外，我们介绍了一个可重用的基准套件，它使我们能够轻松地模拟和测试典型的ClickHouseKeeper使用场景。基于此，我们呈现基准测试结果，突显ClickHouseKeeper在保持性能接近ZooKeeper的同时，使用的内存却只有ZooKeeper的1/46。

动机

现代分布式系统需要一个共享且可靠的信息存储库和一致性系统，以协调和同步分布式操作。对于ClickHouse，最初选择了ZooKeeper。它广泛被使用，证明是可靠的，并提供了简单而强大的API，合理的性能。

然而，对于ClickHouse，不仅性能，而且资源效率和可扩展性一直是首要任务。ZooKeeper作为一个Java生态系统项目，不能很好地优雅地融入我们的C++代码库，而且随着我们以越来越高的规模使用它，我们开始遇到资源使用和运维挑战。为了克服ZooKeeper的这些缺点，考虑到我们的项目需要解决的其他要求和目标，我们从零开始构建了ClickHouseKeeper。

ClickHouseKeeper是ZooKeeper的替代方案，具有完全兼容的客户端协议和相同的数据模型。除此之外，它还提供以下好处：

• 更容易的设置和操作：ClickHouseKeeper是用C++实现的，而不是Java，因此可以嵌入到ClickHouse中或独立运行
  
• 由于更好的压缩，快照和日志占用的磁盘空间要少得多
  
• 默认数据包和节点数据大小没有限制（在ZooKeeper中是1MB）
  
• 没有ZXID溢出问题（在ZooKeeper中，它在每20亿次事务时强制重新启动）
  
• 因为使用了更好的分布式一致性协议，在网络分区后能更快地恢复。
  
• 额外的一致性保证：ClickHouseKeeper提供与ZooKeeper相同的一致性保证-线性可写，以及相同会话内严格的操作顺序。此外，通过设置quorum_reads，ClickHouseKeeper还提供线性读取。
  
• ClickHouseKeeper对于相同数量的数据更节约资源，使用的内存更少（稍后在本博客中我们将证明这一点）
  

ClickHouseKeeper的开发始于2021年2月，作为ClickHouse服务中的嵌入式服务。同一年，引入了独立模式，并添加了Jepsen测试-每6小时，我们运行带有多种不同工作流和失败场景的自动化测试，以验证一致性机制的正确性。

在撰写本博客时，ClickHouseKeeper已经在生产环境中运行了一年半以上，并且自2022年5月首次进行私人预览以来，已经在我们自己的ClickHouseCloud中大规模部署。

在博客的其余部分，我们有时将ClickHouseKeeper简称为“Keeper”，因为我们内部经常这样称呼它。

在ClickHouse中的使用

通常，任何需要在多个ClickHouse服务器之间保持一致性的事物都依赖于Keeper：

• Keeper为自建的shared-nothingClickHouse集群中的数据复制提供协调系统
  
• Mergetree引擎系列的复制表的自动插入去重是基于存储在Keeper中的block-hash-sums实现的
  
• Keeper为part名称（基于顺序块编号）提供一致性，并为将part合并和mutation分配给特定集群节点提供一致性
  
• Keeper在KeeperMap表引擎的后台中使用，该引擎允许您使用Keeper作为一致的键值存储，具有线性可写和顺序一致的读取
  

• 利用这一点在ClickHouse上实现任务调度队列的应用程序
  
• KafkaConnectSink使用此表引擎作为可靠的状态存储，以实现精确一次交付保证
  

• Keeper持续跟踪S3Queue表引擎中消费的文件
  
• 复制的Database引擎将所有元数据存储在Keeper中
  
• Keeper用于与ONCLUSTER子句一起协调备份
  
• UDF可以存储在Keeper中
  
• 访问控制信息可以存储在Keeper中
  
• Keeper用作ClickHouseCloud中所有元数据的共享中央存储
  

观测Keeper

在接下来的几节中，为了观测（并稍后在基准测试中模拟）ClickHouseCloud与Keeper的一些交互，我们将一个月的WikiStat数据集加载到一个具有3个节点的ClickHouseCloud服务中。每个节点有30个CPU核心和120GB的RAM。每个服务都使用自己专用的ClickHouseKeeper服务，包括3个服务器，每个Keeper服务器有3个CPU核心和2GBRAM。

下图说明了这个数据加载场景：

①数据加载

通过一个数据加载，在大约100秒内，我们从大约740个压缩文件中（一个文件表示一天中的一个特定小时）并行加载了约46亿行数据到所有三个ClickHouse服务器。单个ClickHouse服务器的内存使用峰值约为107GB：

0 rows inset. Elapsed:101.208 sec. Processed 4.64 billion rows,40.58 GB (45.86 million rows/s.,400.93 MB/s.)Peak memory usage:107.75 GiB.

②Part创建

为了存储数据，3个ClickHouse服务器共同在对象存储中创建了240个part。每个初始part的平均行数约为1934万行。平均大小约为100MiB，插入的总行数为46亿：

┌─parts──┬─rows_avg──────┬─size_avg───┬─rows_total───┐
│ 240.00 │ 19.34 million │ 108.89 MiB │ 4.64 billion │
└────────┴───────────────┴────────────┴──────────────┘

由于我们的数据加载利用了s3Cluster表函数，part的创建均匀分布在ClickHouseCloud服务的3个ClickHouse服务器上：

┌─n─┬─parts─┬─rows_total───┐
│ 1 │ 86.00 │ 1.61 billion │
│ 2 │ 76.00 │ 1.52 billion │
│ 3 │ 78.00 │ 1.51 billion │
└───┴───────┴──────────────┘

③part合并

在数据加载过程中，ClickHouse在后台执行了1706次part合并：

┌─merges─┐
│   1706 │
└────────┘

④Keeper交互

ClickHouse完全将数据和元数据的存储从服务器中分离。所有数据都存储在共享对象存储中，所有元数据存储在Keeper中。当ClickHouse服务器将新的part写入对象存储（参见②上文）或将一些part合并为新的较大part时（参见③上文），则该ClickHouse服务器会使用事务请求多写来更新Keeper中关于新part的元数据。此信息包括part的名称、哪些文件属于该part以及与文件对应的blob在对象存储中的位置。每个服务器都有一个包含元数据子集的本地缓存，并通过Keeper实例通过基于订阅的监视机制自动了解数据更新。

对于我们上述的初始part的创建和后台part的合并，执行了总共约18,000次Keeper请求。其中包括约12,000次多写事务请求（仅包含写子请求）。所有其他请求是读写请求的混合。此外，ClickHouse服务器从Keeper收到了约800次通知：

total_requests:      17705
multi_requests:      11642
watch_notifications: 822

我们可以看到这些请求如何相当均匀地从所有三个ClickHouse节点发送和接收监视通知：

┌─n─┬─total_requests─┬─multi_requests─┬─watch_notifications─┐
│ 1 │           5741 │           3671 │                 278 │
│ 2 │           5593 │           3685 │                 269 │
│ 3 │           6371 │           4286 │                 275 │
└───┴────────────────┴────────────────┴─────────────────────┘

以下两张图可视化了在数据加载过程中这些Keeper请求：

我们可以看到大约70%的Keeper请求是多写事务。

请注意，Keeper请求的数量可能会根据ClickHouse集群大小、写入设置和数据大小而变化。我们简要演示了这三个因素如何影响生成的Keeper请求数量。

ClickHouse集群大小

如果我们使用10台服务器而不是3台服务器并行加载数据，我们可以将数据写入速度提高3倍多（使用SharedMergeTree）：

0 rows in set. Elapsed: 33.634 sec. Processed 4.64 billion rows, 40.58 GB (138.01 million rows/s., 1.21 GB/s.)
Peak memory usage: 57.09 GiB.

更多服务器的数量会产生超过3倍的Keeper请求量：

total_requests:      60925
multi_requests:      41767
watch_notifications: 3468

写入设置

对于我们最初使用3台ClickHouse服务器运行的数据加载，我们配置了每个初始part的最大大小为~25百万行，以加快写入速度，代价是更高的内存使用。如果我们改为使用默认值~1百万行每个初始part运行相同的数据加载，则数据加载速度较慢，但每个ClickHouse服务器的主要内存使用量减少了约9倍：

0 rows in set. Elapsed: 121.421 sec. Processed 4.64 billion rows, 40.58 GB (38.23 million rows/s., 334.19 MB/s.)
Peak memory usage: 12.02 GiB.

并且创建的初始part数量从240个增加到了约4千个：

┌─parts─────────┬─rows_avg─────┬─size_avg─┬─rows_total───┐
│ 4.24 thousand │ 1.09 million │ 9.20 MiB │ 4.64 billion │
└───────────────┴──────────────┴──────────┴──────────────┘

这导致了更多的part合并：

┌─merges─┐
│   9094 │
└────────

我们得到了更多的Keeper请求（约147k而不是约17k）：

total_requests:      147540
multi_requests:      105951
watch_notifications: 7439

数据大小

同样，如果我们加载更多的数据（使用默认值~1百万行每个初始part），例如六个月的WikiStat数据，则我们的服务将获得更多约24k个初始part：

─parts──────────┬─rows_avg─────┬─size_avg─┬─rows_total────┐
│ 23.75 thousand │ 1.10 million │ 9.24 MiB │ 26.23 billion │
└────────────────┴──────────────┴──────────┴───────────────┘

这导致了更多的合并：

┌─merges─┐
│  28959 │
└────────┘

从而产生了约680k的Keeper请求：

total_requests:      680996
multi_requests:      474093
watch_notifications: 32779

Keeper基准测试

我们开发了一个名为`keeper-bench-suite`的基准测试套件，用于基准测试上述探讨的ClickHouse与Keeper的典型交互。为此，`keeper-bench-suite`允许模拟来自由N台（例如3台）服务器组成的ClickHouse集群的并行Keeper工作负载：

我们利用了`keeper-bench`，这是一个用于基准测试Keeper或任何与ZooKeeper兼容的系统的工具。借助这个基础构建块，我们可以模拟并基准测试从N台ClickHouse服务器发送的典型并行Keeper流量。此图显示了`KeeperBenchSuite`的完整架构，它使我们能够轻松设置和基准测试任意Keeper工作负载场景：

我们使用AWSEC2实例作为基准服务器，执行一个Python脚本，该脚本执行以下操作：

①通过启动3个适当配置的EC2实例（例如m6a.4xlarge），每个实例运行一个KeeperDocker容器和两个包含cAdvisor和Redis（cAdvisor需要）的容器，从而设置和启动一个3节点的Keeper集群。

②启动`keeper-bench`并使用预配置的工作负载配置。

③每秒从每个cAdvisor和Keeper的Prometheus端点获取数据。

④将获取的指标及时间戳写入ClickHouseCloud服务中的两个表中，这是通过SQL查询和Grafana仪表板方便地分析指标的基础。

请注意，ClickHouseKeeper和ZooKeeper直接提供Prometheus端点。目前，这些端点的重叠部分很小，并且通常提供非常不同的指标，特别是在内存和CPU使用方面。因此，我们选择了基于cAdvisor的基本容器指标。此外，将Keeper运行在Docker容器中使我们可以轻松更改为Keeper提供的CPU核心数和内存大小。

配置参数

Keeper的大小

我们使用不同的Docker容器大小运行ClickHouseKeeper和ZooKeeper的基准测试。例如，1CPU核心+1GBRAM，3CPU核心+1GBRAM，6CPU核心+6GBRAM。

客户端数量和请求数

对于每个Keeper大小，我们使用`keeper-bench`的并发设置模拟不同数量的客户端（例如，ClickHouse服务器）并行发送请求到Keeper：例如3、10、100、500、1000。

从这些模拟的客户端中，为了模拟短时和长时运行的Keeper会话，我们使用`keeper-bench`的迭代设置向Keeper发送总计介于1万和~1000万请求。这旨在测试组件的内存使用量是否随时间变化。

工作负载

我们模拟了一个典型的ClickHouse工作负载，其中包含约1/3的写入和删除操作，以及约2/3的读取。这反映了一种情景，其中一些数据被写入、合并，然后进行查询。很容易定义和基准测试其他工作负载。

测量指标

Prometheus端点

我们使用cAdvisor的Prometheus端点测量：

• 主内存使用量（container_memory_working_set_bytes）
  
• CPU使用量（container_cpu_usage_seconds_total）
  

我们使用ClickHouseKeeper和ZooKeeper的Prometheus端点测量附加的（所有可用的）KeeperPrometheus端点指标值。例如，对于ZooKeeper，有许多JVM特定的指标（堆大小和使用情况、垃圾回收等）。

运行时间

我们还根据每次运行的最小和最大时间戳测量Keeper处理所有请求的运行时间。

结果

我们使用`keeper-bench-suite`比较了ClickHouseKeeper和ZooKeeper在我们的工作负载下的资源消耗和运行时间。我们每个基准配置运行10次，并将结果存储在ClickHouseCloud服务中的两个表中。我们使用SQL查询生成了三个表格化的结果表：

• 平均值
  
• 95分位数
  
• 99分位数
  

这些结果的列在此处描述。

我们使用ClickHouseKeeper23.5和ZooKeeper3.8（带有捆绑的OpenJDK11）进行所有运行。请注意，我们不在这里列出三个表格化的结果，因为每个表格包含216行。您可以通过上面的链接查看结果。

示例结果

在这里，我们展示了两个图表，其中过滤了处理相同大小请求的3个模拟客户端（ClickHouse服务器）并行发送的情况下，两个Keeper版本使用3个CPU核心和2GBRAM运行的99分位数结果。这些可视化的表格结果在此处。

内存使用

我们可以看到，对于我们模拟的工作负载，ClickHouseKeeper对于相同数量的处理请求，始终比ZooKeeper使用更少的主内存。例如，对于处理由3个模拟ClickHouse服务器并行发送的640万个请求的基准运行③，ClickHouseKeeper的主内存使用量比运行④中的ZooKeeper少约46倍。

对于ZooKeeper，我们使用了1GiB的JVM堆大小配置（JVMFLAGS：-Xmx1024m-Xms1024m）进行所有主要运行（①、②、③），这意味着对于这些运行，已提交的JVM内存（保留的堆和非堆内存保证可供Java虚拟机使用）大小为~1GiB（请参见上图中透明的灰色条，显示使用了多少）。除了Docker容器的内存使用（蓝条）之外，我们还测量了实际在提交的JVM内存内使用的（堆和非堆）JVM内存量（粉条）。运行JVM本身有一些轻微的容器内存开销（蓝条和粉条的差异）。然而，实际使用的JVM内存仍然一致显着大于ClickHouseKeeper的总体容器内存使用。

此外，我们可以看到ZooKeeper在运行③期间使用了完整的1GiB的JVM堆大小。我们进行了额外的运行④，使用增加的2GiBJVM堆大小进行了额外的运行，导致ZooKeeper使用了其2GiBJVM堆的1.56GiB，运行时间与ClickHouseKeeper的运行③相匹配。我们在下一个图表中展示了所有运行的运行时间。我们在表格结果中看到，在ZooKeeper运行期间发生了几次（主要）垃圾回收。

运行时间和CPU使用

下图展示了在前一个图表中讨论的运行的运行时间和CPU使用情况（两个图表中的圆圈数字是对齐的）：

ClickHouseKeeper的运行时间与ZooKeeper的运行时间非常相近。尽管ClickHouseKeeper使用的主内存明显较少（请参见前一个图表），而且CPU使用率较低。

扩展Keeper

我们观察到ClickHouseCloud在与Keeper交互时经常使用多写事务。我们更深入地了解ClickHouseCloud与Keeper的交互，勾勒出ClickHouse服务器使用的两个主要场景中的这些Keeper事务。

自动插入去重

在上面勾画的场景中，服务器-2①以块的形式处理插入到表中的数据。对于当前的块，服务器-2②将数据写入对象存储中的新part，并使用Keeper多写事务在Keeper中存储有关新part的元数据，例如，part文件对应的blob存储在对象存储中的位置。在存储此元数据之前，事务首先尝试将步骤①中处理的块的哈希总和存储在Keeper中的去重日志znode中。如果相同的哈希总和值已经存在于去重日志中，则整个事务失败（被回滚）。此外，步骤②中的数据part会被删除，因为该part中包含的数据已经在过去插入过了。这种自动插入去重使得ClickHouse插入变得幂等，因此容错，允许客户端重试插入而无需担心数据重复。成功后，事务触发子监视器，④所有订阅了part元数据znodes事件的ClickHouse服务器都会被Keeper自动通知新条目。这导致它们从Keeper中获取元数据更新到它们的本地元数据缓存中。

分配part合并给服务器

当server-2决定将一些part合并成一个较大的part时，那么该服务器①使用Keeper事务将待合并的part标记为已锁定（以防止其他服务器将其合并）。接下来，服务器-2②将这些part合并成一个新的更大的part，③使用另一个Keeper事务存储有关新part的元数据，这触发监视器④通知所有其他服务器有关新元数据条目。

请注意，上述场景只有在这样的Keeper事务被Keeper原子地和顺序地执行时才能正确工作。否则，同时并行发送相同数据的两个ClickHouse服务器在去重日志中可能无法找到数据的哈希总和，从而导致在对象存储中出现数据重复。或者多个服务器将合并相同的part。为防止这种情况发生，ClickHouse服务器依赖于Keeper的一切或无事务和其可线性化写入保证。

线性一致性与多核处理

ZooKeeper和ClickHouseKeeper中的共识算法，分别是ZAB和Raft，都确保多个分布式服务器可以可靠地达成相同的信息，例如上面的示例中允许合并哪些part。

ZAB是ZooKeeper的专用共识机制，至少从2008年以来一直在开发中。