隆重推出全新官方 ClickHouse Kafka 连接器

作为一家扎根于开源的公司，我们相信让用户有机会协作和尝试新功能。本着这种精神，我们宣布发布开源 Kafka Connect Sink for ClickHouse 的 Beta 版本，并邀请我们的社区开始测试并提供关于设计和可能的改进领域的反馈。下面我们将讨论开发此连接器的原因，以及我们如何提议解决精确一次交付语义的问题。

Apache Kafka 是一个无处不在的开源分布式事件流平台，数千家公司使用它进行高性能数据管道、流分析、数据集成和关键任务应用程序。

ClickHouse 和 Kafka 是互补的，用户需要将基于 Kafka 的数据插入 ClickHouse 以进行大规模分析。现有的解决方案已经存在用于解决这个问题，那么让我们深入探讨一下我们为什么要构建另一个连接器。

在探索当前 ClickHouse-Kafka 的情况之前，让我们回顾一下精确一次交付的含义，这与其他选项有何关系，以及每种选项可能适用的情况。

Apache Kafka 支持三种消息交付语义，此处按实现复杂程度升序列出

• 最多一次：一条消息最多只交付一次，或者根本不交付。
  这种情况优先考虑性能和吞吐量，而不是数据一致性，通过保持最小的消息传递开销。它可能适用于大规模部署的日志和指标收集，例如，在物联网领域。在这种情况下，即使缺少一些事件，仍然可以从统计上显着的观察结果中得出一些结论。
• 至少一次：一条消息可以交付一次或多次，但保证永远不会丢失。这种方法代表了一个有趣的中间地带，其中对重复项的容忍度降低了操作复杂性。然后，数据存储将需要在查询层或通过提供去重功能来弥补，以确保重复项不会损害业务结论。这是迄今为止跨越许多用例（如可观测性）最常见的方法。
• 精确一次：一条消息将始终只交付一次。这种方法对于业务关键型应用（如金融分析）至关重要，在这些应用中，准确性不能妥协，并且接收系统无法去重。它带来了显着的操作开销，因为需要跟踪许多移动组件，以便能够从部分状态恢复摄取。

ClickHouse 支持多种技术来实现从 Kafka 的摄取，每种技术都有其各自的优点和缺点。

Kafka 表引擎提供了原生的 ClickHouse 集成，可用于将数据从 Kafka 插入到 ClickHouse 中，反之亦然。作为另一个表引擎，其架构简洁性吸引了入门用户，因为不需要额外的组件。然而，它确实有一些缺点。调试错误和内省当前行为可能具有挑战性，尽管我们计划改进。此外，它会给您的 ClickHouse 集群带来额外的负载，并要求用户在常规插入和查询负载的上下文中考虑这一点。因此，在架构上，我们经常看到用户希望分离这些任务。最重要的是，作为一种拉取架构，它要求您的 ClickHouse 集群具有与 Kafka 的双向连接 - 如果存在网络隔离，例如，ClickHouse 在云端，而 Kafka 是自管理的，您可能会因为合规性和安全原因而犹豫是否允许双向连接。

对于那些需要推送架构和/或想要分离架构关注点的用户，Vector 和 Kafka Connect 也是现有的选项。

Vector 使用其 Kafka 输入和ClickHouse 输出，是一个出色的解决方案，但它是一个专注于可观测性的工具，不适用于所有用例。

Kafka Connect 是 Apache Kafka 的一个免费、开源组件，充当集中式数据中心，用于数据库、键值存储、搜索索引和文件系统之间的简单数据集成。此框架支持两种类型的连接器：接收器（从 Kafka 到目标）和源（从源到 Kafka）。对于 ClickHouse，可以使用 HTTP 和 JDBC 连接器与 ClickHouse 集成。这些又带来了挑战。JDBC 连接器既是接收器又是源，并在 社区许可证下分发。但是，目前不支持 ClickHouse 方言，这意味着它仅适用于基本的 ClickHouse 类型，例如 Int32。相反，HTTP Sink 通过 JSON 和 ClickHouse HTTP 接口，支持所有类型，但它是商业许可的。

最重要的是，以上所有方案最多只能提供至少一次交付。

最终，我们希望为用户提供一个基于推送的连接器，它支持所有 ClickHouse 类型和精确一次交付语义。不幸的是，至少一次交付通常是实现 Kafka 连接器的基本设计决策。虽然我们可以通过增强现有连接器（例如，向 JDBC sink 添加方言支持）来解决早期的一些挑战，但这并不能解决精确一次的要求。

当面对以上所有挑战时，我们决定构建一个新的连接器。我们需要决定是创建一个单独的组件还是使用现有的框架。鉴于 Kafka Connect 框架的普及及其在 MSK 和 Confluent Cloud 中的支持，这两者在我们的用户中越来越受欢迎，我们决定为 Kafka Connect 框架构建一个新的连接器，并为 ClickHouse Cloud 提供一流的支持。

在我们的要求中，实现精确一次交付语义提出了最大的挑战。在我们讨论我们如何提议实现这一目标之前，让我们回顾一下至少一次语义有时会导致重复的原因。这些原因通常是连接器特定的，但通常分为两种模式，都与如何从 Kafka 消费消息以及跟踪消息队列上当前位置（偏移量）的方式有关

• 消费者处理一条 Kafka 消息，将其发送到 ClickHouse，并在崩溃并丢失其内存偏移量之前提交偏移量。在这种情况下，Kafka 已配置为自动提交偏移量（默认），但没有机会提交从消费者收到的偏移量（这是周期性的）。消费者重新启动，因此，它从上次提交的偏移量交付消息，而它已经消费过这些消息。
• 消费者使用提交 API（禁用自动提交）并负责在 Kafka 中提交偏移量。它处理一条 Kafka 消息，将其发送到 ClickHouse，但在将偏移量提交到 Kafka 之前崩溃。一旦重新启动，Kafka 从上次偏移量交付消息，导致重复项被发送到 ClickHouse。

请注意，这些情况假设偏移量在 Kafka 中跟踪。确切的原因通常取决于集成、偏移量提交策略和使用的 API。更多阅读此处。

解决此问题的常用方法是在目标数据存储中管理您的偏移量。这可能涉及多种方法，通常取决于目标数据存储的属性。例如，如果数据存储提供 ACID 事务，则连接器可以将偏移量与消息一起提交。在没有 ACID 事务的情况下，使用不同的存储来存储偏移量的两阶段提交可能是可行的。通常，这些方法会产生开销并降低吞吐量。

在考虑解决方案时，我们希望找到一种涉及最少依赖项、架构简单且利用 ClickHouse 现有功能的方案。我们的完整设计可以在此处找到。当我们增加测试并转向全面可用性时，我们欢迎反馈。

简而言之，我们实现精确一次交付的设计依赖于利用 ClickHouse 的插入去重功能，确保我们始终使用状态机为插入制定一致的批次，并使用 Kafka Connect API 开发连接器，以便我们在发生故障时始终接收重复项。这种方法通过保证重复记录的去重来增强 Kafka Connect 的至少一次语义，从而实现精确一次交付。

ClickHouse Keeper 为与 ClickHouse 集群协调系统关联的数据提供强一致性存储，并且是允许 ClickHouse 作为分布式系统运行的基础。这支持诸如数据复制、分布式 DDL 查询执行、领导者选举和服务发现等服务。ClickHouse Keeper 与 ZooKeeper 兼容，ZooKeeper 是 ClickHouse 中用于此功能的旧组件。与 Zookeeper 类似，ClickHouse Keeper 支持写入的线性化和读取的顺序一致性。但是，它比 Zookeeper 具有明显的优势，主要是压缩日志、轻量级安装、更小的内存消耗（没有 JVM），甚至可以选择读取的线性化。当需要高度一致的存储时，这些属性非常适合持久化少量数据。

我们提出的连接器设计要求连接器在具有顺序一致性和线性化写入的强一致性存储中存储状态。最初，我们考虑了 ClickHouse，但由于几个原因很快就放弃了。首先，ClickHouse 默认不是强一致性的，并且仅提供最终一致性复制。但是，通过仔细配置，您可以确保 线性化插入和SELECT 的顺序一致性对于 复制表。但是，这种 ClickHouse 配置增加了显着的插入延迟，主要是因为与 ClickHouse Keeper 的通信增加以协调写入和后续数据复制。这种设计有效地添加了一个冗余组件和不必要的开销 - ClickHouse 表存储。鉴于我们只需要存储最少的状态，直接使用 ClickHouse Keeper 似乎是满足这些要求的完美解决方案。

这种方法的挑战在于，此组件通常不在集群中公开。例如，它在 ClickHouse Cloud 中未公开，对其访问和使用应谨慎控制，以免影响集群操作和稳定性。与 ClickHouse 核心团队合作，我们决定通过表引擎 - KeeperMap 引擎，以受控方式公开 ClickHouse Keeper（对于需要线性化插入和顺序一致性的情况）。这为我们提供了一种集成且轻量级的方式来存储我们的状态。

请注意，您可以使用内存模式在没有 KeeperMap 的情况下测试连接器。这仅用于测试，并且在发生故障时不做任何精确一次的保证。

在接下来的几个月中，我们计划在各种故障场景下广泛测试连接器。一旦我们对设计和实现充满信心，并收集到您（我们的用户）的反馈，该连接器将全面可用。

尽管处于 Beta 阶段，但该连接器已经功能丰富，支持大多数 ClickHouse 类型（包括数组和 Map），并允许在有或没有模式的情况下插入数据。在没有模式的情况下，数据在插入之前首先转换为 JSON。对于具有模式的数据，我们使用 RowBinary 格式以获得最佳性能。这两种方法都使用 ClickHouse HTTP 接口，并且我们持续针对 ClickHouse Cloud 进行测试。

最后，除了 一些当前限制之外，我们计划添加对删除的支持，允许将 Redis 用作状态存储和JSON 类型。

在这篇博文中，我们探讨了我们如何以及为什么为 ClickHouse 构建了一个新的 Kafka 连接器。我们解释了我们为实现精确一次交付语义而提出的方法，以及这如何克服现有解决方案的局限性。试用新的连接器，我们欢迎反馈！