ClickHouse Keeper：用 C++ 编写的 ZooKeeper 替代方案

ClickHouse 是用于实时应用和分析的最快速、资源效率最高的开源数据库。作为其组件之一，ClickHouse Keeper 是 ZooKeeper 的快速、更资源高效且功能丰富的替代方案。这个开源组件提供高度可靠的元数据存储，以及协调和同步机制。它最初是为与 ClickHouse 一起使用而开发的，当 ClickHouse 作为分布式系统部署在自管理设置或像 CloudHouse Cloud 这样的托管产品中时。但是，我们相信更广泛的社区可以从这个项目中获益，用于更多的用例。

在这篇文章中，我们描述了 ClickHouse Keeper 的动机、优势和开发过程，并预览了我们接下来计划的改进。此外，我们还介绍了一个可重用的基准测试套件，它使我们能够轻松地模拟和测试典型的 ClickHouse Keeper 使用模式。基于此，我们展示了基准测试结果，突出表明 ClickHouse Keeper 在保持接近 ZooKeeper 性能的同时，对于相同的数据量，内存使用量最多可减少 46 倍。

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

然而，不仅性能，资源效率和可扩展性一直是 ClickHouse 的首要 优先事项。ZooKeeper 作为一个 Java 生态系统项目，与我们主要的 C++ 代码库不太协调，并且随着我们在越来越高的规模上使用它，我们开始遇到资源使用和操作方面的挑战。为了克服 ZooKeeper 的这些缺点，我们从头开始构建了 ClickHouse Keeper，同时考虑了我们的项目需要解决的其他需求和目标。

ClickHouse Keeper 是 ZooKeeper 的直接替代品，具有完全兼容的客户端协议和相同的数据模型。除此之外，它还提供以下优势：

• 更简易的设置和操作：ClickHouse Keeper 是用 C++ 而不是 Java 实现的，因此 可以 嵌入在 ClickHouse 中运行或独立运行
• 由于更好的压缩，快照和日志消耗的磁盘空间更少
• 默认数据包和节点数据大小没有限制（在 ZooKeeper 中 是 1 MB）
• 没有 ZXID 溢出 问题（在 ZooKeeper 中每 2B 事务就会强制重启）
• 由于使用了更好的分布式共识协议，网络分区后恢复速度更快
• 额外的 一致性 保证：ClickHouse Keeper 提供与 ZooKeeper 相同的一致性保证 - 线性化 写入加上在同一 会话 内操作的严格排序。此外，并且可选地（通过 quorum_reads 设置），ClickHouse Keeper 提供线性化读取。
• ClickHouse Keeper 资源效率更高，并且对于相同的数据量使用更少的内存（我们将在本文后面演示这一点）

ClickHouse Keeper 的开发 开始于 2021 年 2 月，作为 ClickHouse 服务器中的嵌入式服务。同年，引入 了独立模式，并且 Jepsen 测试被 添加 - 每 6 小时，我们运行自动化的 测试，包含几种不同的工作流程和故障场景，以验证共识机制的正确性。

在撰写此博客时，ClickHouse Keeper 已经生产就绪 超过 一年半，并且自 2022 年 5 月首次私有预览发布以来，已在我们自己的 ClickHouse Cloud 中大规模部署。

在本文的其余部分，我们有时将 ClickHouse Keeper 简称为 “Keeper”，就像我们内部经常称呼它一样。

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

• Keeper 为自管理的 无共享架构 ClickHouse 集群 中的数据 复制 提供协调系统
• 对于 mergetree 引擎系列的复制表，自动 插入去重 基于存储在 Keeper 中的块哈希和 值
• Keeper 为 part 名称（基于顺序 块 编号）以及将 part 合并 和 mutation 分配给特定的集群节点提供共识
• Keeper 在 KeeperMap 表引擎 的底层使用，允许您将 Keeper 用作具有线性化写入和顺序一致性读取的一致性键值存储
  • 阅读 关于一个应用程序利用此功能在 ClickHouse 之上实现任务调度队列的文章
  • Kafka Connect Sink 使用此表引擎作为可靠的 状态存储，用于 实现 精确一次交付 保证
• Keeper 跟踪 S3Queue 表引擎 中已消费的文件
• Replicated Database 引擎 将所有元数据存储在 Keeper 中
• Keeper 用于协调使用 ON CLUSTER 子句的 备份
• 用户定义的函数 可以 存储 在 Keeper 中
• 访问控制 信息可以 存储 在 Keeper 中
• Keeper 被用作 ClickHouse Cloud 中所有元数据的共享中心存储

在以下章节中，为了观察（并在稍后的基准测试中建模）ClickHouse Cloud 与 Keeper 的一些交互，我们将来自 WikiStat 数据集的一个月的数据加载到一个 表 中，该表位于具有 3 个节点的 ClickHouse Cloud 服务 中。每个节点有 30 个 CPU 核心和 120 GB RAM。每个服务都使用其自己的专用 ClickHouse Keeper 服务，该服务由 3 个服务器组成，每个 Keeper 服务器具有 3 个 CPU 核心和 2 GB RAM。

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

通过数据加载 查询，我们使用所有三个 ClickHouse 服务器在 并行 模式下，在大约 100 秒内从大约 740 个压缩文件中加载了约 46.4 亿行（一个文件代表特定日期的特定小时）。单个 ClickHouse 服务器上的峰值主内存使用量约为 107 GB

0 rows in set. 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.

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

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

由于我们的数据加载查询使用了 s3Cluster 表函数，因此初始 part 的创建 被 均匀地分配到我们 ClickHouse Cloud 服务的 3 个 ClickHouse 服务器上

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

在数据加载期间，在后台，ClickHouse 分别 执行了 1706 次 part 合并

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

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

对于我们前面提到的初始 part 创建和后台 part 合并，总共 执行了 约 1.8 万个 Keeper 请求。这包括约 1.2 万个 multi-write 事务请求（仅包含 write-subrequests）。所有其他请求都是读请求和写请求的混合。此外，ClickHouse 服务器从 Keeper 收到了约 800 个 watch 通知

total_requests:      17705
multi_requests:      11642
watch_notifications: 822

我们可以 看到 这些请求是如何发送的，以及 watch 通知是如何从所有三个 ClickHouse 节点相当均匀地接收的

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

以下两张图表可视化了数据加载过程中这些 Keeper 请求 的情况： 我们可以看到，约 70% 的 Keeper 请求是 multi-write 事务。

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

如果我们使用 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.

服务器数量越多，生成的 Keeper 请求数量也超过 3 倍

total_requests:      60925
multi_requests:      41767
watch_notifications: 3468

对于我们使用 3 个 ClickHouse 服务器运行的 原始 数据加载，我们配置了每个初始 part 的最大大小约为 2500 万行，以加快摄取速度，但代价是更高的内存使用率。相反，如果我们 运行 相同的数据加载，并使用每个初始 part 约 100 万行的 默认 值，则数据加载速度较慢，但每个 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.

并且 创建了 约 4 千个而不是 240 个初始 part

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

这 导致 更多的 part 合并

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

我们 获得 了更多的 Keeper 请求（约 14.7 万个而不是约 1.7 万个）

total_requests:      147540
multi_requests:      105951
watch_notifications: 7439

类似地，如果我们 加载 更多数据（使用每个初始 part 约 100 万行的默认值），例如来自 WikiStat 数据集的六个月的数据，那么我们的服务将获得约 2.4 万个初始 part

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

这 导致 更多合并

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

导致 约 68 万个 Keeper 请求

total_requests:      680996
multi_requests:      474093
watch_notifications: 32779

我们开发了一个名为 keeper-bench-suite 的基准测试套件，用于测试上面探讨的 ClickHouse 与 Keeper 的典型交互。为此，keeper-bench-suite 允许模拟来自由 N 个（例如 3 个）服务器组成的 ClickHouse 集群的并行 Keeper 工作负载： 我们正在借助 keeper-bench，这是一个用于基准测试 Keeper 或任何兼容 ZooKeeper 的系统的工具。借助该构建块，我们可以模拟和测试来自 N 个 ClickHouse 服务器的典型并行 Keeper 流量。此图显示了 Keeper Bench Suite 的完整架构，它使我们能够轻松设置和测试任意 Keeper 工作负载场景： 我们正在使用 AWS EC2 实例作为基准测试服务器，用于执行 Python 脚本，该脚本 ① 通过启动 3 个合适的（例如，m6a.4xlarge）EC2 实例来设置和启动 3 节点 Keeper 集群，每个实例运行一个 Keeper docker 容器和两个包含 cAdvisor 和 Redis（cAdvisor 需要）的容器，用于监控本地 Keeper 容器的资源使用情况

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

③ 每 1 秒抓取每个 cAdvisor 和 Keeper 的 Prometheus 端点

④ 将抓取的指标（包括时间戳）写入 ClickHouse Cloud 服务中的两个 表 中，这是通过 SQL 查询方便地分析指标和 Grafana 仪表板的基础

请注意，ClickHouse Keeper 和 ZooKeeper 都直接提供 Prometheus 端点。目前，这些端点只有很小的重叠，并且通常给出非常不同的指标，这使得它们难以比较，尤其是在内存和 CPU 使用率方面。因此，我们选择了基于 cAdvisor 的其他基本容器指标。此外，在 docker 容器中运行 Keeper 使我们能够轻松更改提供给 Keeper 的 CPU 核心数和 RAM 大小。

我们使用不同大小的 docker 容器运行 ClickHouse Keeper 和 ZooKeeper 的基准测试。例如，1 个 CPU 核心 + 1 GB RAM，3 个 CPU 核心 + 1 GB RAM，6 个 CPU 核心 + 6 GB RAM。

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

从这些模拟客户端中的每一个，为了模拟短时和长时间运行的 Keeper 会话，我们（使用 keeper-bench 的 iterations 设置）向 Keeper 发送总共 1 万到约 1 千万个请求。这旨在测试任一组件的内存使用率是否随时间变化。

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

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

• 主内存使用率 (container_memory_working_set_bytes)
• CPU 使用率 (container_cpu_usage_seconds_total)

我们使用 ClickHouse Keeper 和 ZooKeeper 的 Prometheus 端点来测量其他（所有可用的）Keeper Prometheus 端点指标值。例如，对于 ZooKeeper，许多 JVM 特定的指标（堆大小和使用率、垃圾回收等）。

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

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

• 平均值
• 第 95 个百分位数
• 第 99 个百分位数

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

我们对所有运行使用了 ClickHouse Keeper 23.5 和 ZooKeeper 3.8（带有捆绑的 OpenJDK 11）。请注意，我们在此处不打印这三个表格结果，因为每个表包含 216 行。您可以通过关注上面的链接来查看结果。

在这里，我们展示了两个图表，我们在其中 过滤了 第 99 个百分位数结果，用于 Keeper 版本都以 3 个 CPU 核心和 2 GB RAM 运行，并行处理来自 3 个模拟客户端（ClickHouse 服务器）的相同请求大小的行。这些可视化的表格结果在 此处。

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

对于 ZooKeeper，我们对所有主要运行（①、②、③）使用了 1GiB JVM 堆大小配置 (JVMFLAGS: -Xmx1024m -Xms1024m)，这意味着这些运行的提交的 JVM 内存（保留堆和非堆内存保证可供 Java 虚拟机使用）大小约为 1GiB（有关已使用的量，请参见上图中的透明灰色条）。除了 docker 容器内存使用率（蓝色条）之外，我们还测量了在提交的 JVM 内存中实际使用的（堆和非堆）JVM 内存量（粉色条）。运行 JVM 本身存在一些轻微的容器内存 开销（蓝色条和粉色条之间的差异）。但是，实际使用的 JVM 内存仍然始终明显大于 ClickHouse Keeper 的总体容器内存使用率。

此外，我们可以看到 ZooKeeper 在运行 ③ 中使用了完整的 1 GiB JVM 堆大小。我们对 ZooKeeper 进行了额外的运行 ④，JVM 堆大小增加到 2 GiB，导致 ZooKeeper 使用了其 2 GiB JVM 堆中的 1.56 GiB，运行时得到改善，与 ClickHouse Keeper 的运行 ③ 的运行时相匹配。我们在下一张图表中展示了上述所有运行的运行时。

我们可以在表格结果中看到，在 ZooKeeper 运行期间发生了几次（主要）垃圾回收。

下图可视化了上一张图表中讨论的运行的运行时和 CPU 使用率（带圆圈的数字在两张图表中对齐）： ClickHouse Keeper 的运行时与 ZooKeeper 的运行时非常接近。尽管使用了明显更少的主内存（请参阅上一张图表）和 CPU。

我们 观察到 ClickHouse Cloud 在与 Keeper 的交互中经常使用 multi-write 事务。我们更深入地研究 ClickHouse Cloud 与 Keeper 的交互，以草绘 ClickHouse 服务器使用的此类 Keeper 事务的两个主要场景。

在上面草绘的场景中，服务器 2 ① 处理插入到表中的数据，数据以 块 为单位，按块 处理。对于当前块，服务器 2 ② 将数据写入对象存储中的新数据 part，并且 ③ 使用 Keeper multi-write 事务 来存储有关 Keeper 中新 part 的元数据，例如，与 part 文件对应的 blob 在对象存储中的位置。在存储此元数据之前，事务首先尝试将步骤 ① 中处理的块的哈希和存储在 Keeper 中的 deduplication log znode 中。如果相同的哈希和值 已经 存在于 deduplication log 中，则整个事务 失败（回滚）。此外，步骤 ② 中的数据 part 被删除，因为该 part 中包含的数据过去已经插入过。这种自动插入 去重 使 ClickHouse 插入具有 幂等性，因此具有容错性，允许客户端重试插入而不会冒数据重复的风险。成功后，事务触发子 watch，并且 ④ 所有 订阅 了 part-metadata znodes 事件的 ClickHouse 服务器都会收到 Keeper 的自动通知，告知有新条目。这导致他们从 Keeper 将元数据更新提取到其本地元数据缓存中。

当服务器 2 决定将某些 part 合并为更大的 part 时，服务器 ① 使用 Keeper 事务将要合并的 part 标记为锁定（以防止其他服务器合并它们）。接下来，服务器 2 ② 将 part 合并为一个新的更大的 part，并且 ③ 使用另一个 Keeper 事务来存储有关新 part 的元数据，这会触发 watch ④，通知所有其他服务器有关新的元数据条目。

请注意，只有当 Keeper 原子地且按顺序执行此类 Keeper 事务时，上述场景才能正确工作。否则，两个 ClickHouse 服务器同时并行发送相同的数据可能会同时在 deduplication log 中找不到数据的哈希和，从而导致对象存储中数据重复。或者多个服务器将合并相同的 part。为了防止这种情况，ClickHouse 服务器依赖于 Keeper 的全有或全无 multi-write 事务及其线性化写入保证。

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

ZAB 是 ZooKeeper 的专用共识机制，并且自 2008 年以来至少 开发 至今。

我们选择 Raft 作为我们的共识机制，因为它算法简单且 易于理解，并且在我们 2021 年启动 Keeper 项目时，可以使用轻量级且易于集成的 C++ 库。

然而，所有共识算法都是同构的。对于线性化写入，（依赖）转换和事务内的写入操作必须以严格的顺序逐个处理，无论使用哪种共识算法。假设 ClickHouse 服务器并行向 Keeper 发送事务，并且这些事务是相关的，因为它们写入相同的 znodes（例如，本节开头示例场景中的 deduplication log）。在这种情况下，Keeper 只能通过严格按顺序执行此类事务及其操作来保证和实现线性化： 为此，ZooKeeper 使用单线程请求处理器来实现写入请求处理，而 Keeper 的 NuRaft 实现使用单线程全局队列。

通常，线性化使得垂直（更多 CPU 核心）或水平（更多服务器）扩展写入处理速度变得困难。分析和识别独立事务并并行运行它们是可能的，但目前 ZooKeeper 和 ClickHouse Keeper 均未实现此功能。这张图表（我们在此处过滤了第 99 个百分位的结果）突出了这一点： ZooKeeper 和 ClickHouse Keeper 都在 1、3 和 6 个 CPU 核心上运行，并处理来自 500 个客户端并行发送的 128 万个总请求。

（非线性化）读取请求和辅助任务（管理网络请求、批量处理数据等）的性能在理论上可以通过 ZAB 和 Raft 的 CPU 核心数量进行扩展。我们的基准测试结果普遍表明，ZooKeeper 目前在这方面比 ClickHouse Keeper 做得更好，尽管我们一直在持续改进我们的性能（三个 最近的 例子）。

展望未来，我们看到了扩展 Keeper 以更好地支持我们上面描述的场景的需求。因此，我们正在通过这个项目迈出一大步——为 Keeper 引入多组 Raft 协议。

因为如上所述，扩展非分区（非分片）线性化是不可能的，我们将专注于 多组 Raft，我们在其中 分区存储在 Keeper 中的数据。这使得更多事务彼此独立（在单独的分区上工作）。通过在同一服务器内为每个分区使用单独的 Raft 实例，Keeper 会自动并行执行独立的事务： 通过多 Raft，Keeper 将能够支持具有更高并行读/写需求的工作负载，例如，具有数百个节点的大型 ClickHouse 集群。

听起来很令人兴奋？那么，我们邀请您加入 Keeper 社区。

• 这里 介绍了如何在 ClickHouse 中使用 Keeper
• 要成为 ClickHouse 之外的 Keeper 用户 - 请查看此页面，了解何时使用它或不使用它
• 这里是您发布问题的地方；您可以在 X 上关注我们，并加入我们的聚会和活动。

我们欢迎为 Keeper 代码库做出贡献。请在此处查看我们的路线图 here，并在此处查看我们的贡献者指南 here。

在这篇博文中，我们介绍了 ClickHouse Keeper 的特性和优势 - ClickHouse Keeper 是一款资源高效的开源 ZooKeeper 替代品。我们探索了我们在 ClickHouse Cloud 中对它的使用，并在此基础上，展示了一个基准测试套件和结果，突显出 ClickHouse Keeper 在性能相当的情况下，始终比 ZooKeeper 使用更少的硬件资源。我们还分享了我们的路线图以及您可以参与的方式。我们邀请您合作！