ClickHouse 在实时场景的应用和优化-InfoQ

https://v.qq.com/x/page/q3141p7q5m7.html

视频地址

早期实践

外部事务

在介绍实时场景之前，我先简单讲一下早期的离线数据是如何支持的：

在第一场分享中，技术负责人陈星介绍了 ClickHouse 在字节跳动内部最早支持的两个业务场景，用户行为分析平台和敏捷 BI 平台。这两个平台的数据主要由分析师或者数仓同学产出，以 T+1 的离线指标为主。考虑到 ClickHouse 并不支持事务，为了保障数据的一致性，我们在 ClickHouse 系统外实现了一套外部事务：

数仓同学一般会在 HDFS/Hive 准备好原始数据；数据就绪后，会执行一个基于 Spark 的 ETL 服务，将数据切成 N 份再存回 HDFS（必要的话也会做一些数据的预处理）；再发起 INSERT Query 给 ClickHouse 集群的每一个 shard，将对应的数据文件从 HDFS 中直接导入到 MergeTree 表中，需要注意的是，这里没有把数据写入分布式表（i.e. Distributed table）中；每个节点上的 MergeTree 表写入成功之后，会由外部事务校验整个 集群的数据 是否写入成功：如果部分节点导入失败，外部的导入服务会将部分写入的数据回滚并重新执行导入任务，直到数据完全导入成功，才允许上层的分析平台查询数据。也就是说，当 ClickHouse 中仅有不完整的数据时，外部的查询服务不会查询当天的数据。

除了离线的场景，也有业务方希望执行 INSERT Query 将数据即时地导入 ClickHouse 中，从而能查询到实时的数据。然而，我们曾经出现过由于业务同学高频写入数据，导致文件系统压力过大最后无法正常查询的线上问题。

这里我解释一下直接写入数据的风险：

直接写入的风险

用户写入 ClickHouse 一般有两种选择：分布式表（i.e. Distributed），MergeTree 表：

写入分布式表：

数据写入分布式表时，它会将数据先放入本地磁盘的缓冲区，再异步分发给所有节点上的 MergeTree 表。如果数据在同步给 MergeTree 里面之前这个节点宕机了，数据就可能会丢失；此时如果在失败后再重试，数据就可能会写重。因而，直接将数据写入用分布式表时，不太好保证数据准确性的和一致性。

当然这个分布式表还有其他问题，一般来说一个 ClickHouse 集群会配置多个 shard，每个 shard 都会建立 MergeTree 表和对应的分布式表。如果直接把数据写入分布式表，数据就可能会分发给每个 shard。假设有 N 个节点，每个节点每秒收到一个 INSERT Query，分发 N 次之后，一共就是每秒生成 NxN 个 part 目录。集群 shard 数越多，分发产生的小文件也会越多，最后会导致你写入到 MergeTree 的 Part 的数会特别多，最后会拖垮整个文件的系统。

写入 MergeTree 表：

直接写入 MergeTree 表可以解决数据分发的问题，但是依然抗不住高频写入，如果业务方写入频次控制不好，仍然有可能导致 ClickHouse 后台合并的速度跟不上写入的速度，最后会使得文件系统压力过大。

所以一段时间内，我们禁止用户用 INSERT Query 把数据直接写入到 ClickHouse。

典型案例-推荐系统

业务需求

随着 ClickHouse 支持的业务范围扩大，我们也决定支持一些实时的业务，第一个典型案例是推荐系统的实时数据指标：在字节跳动内部 AB 实验应用非常广泛，特别用来验证推荐算法和功能优化的效果。

最初，公司内部专门的 AB 实验平台已经提供了 T+1 的离线实验指标，而推荐系统的算法工程师们希望能更快地观察算法模型、或者某个功能的上线效果，因此需要一份能够实时反馈的数据作为补充。他们大致有如下需求：

研发同学有 debug 的需求，他们不仅需要看聚合指标，某些时间还需要查询明细数据。
推荐系统产生的数据，维度和指标多达几百列，而且未来可能还会增加。
每一条数据都命中了若干个实验，使用 Array 存储，需要高效地按实验 ID 过滤数据。
需要支持一些机器学习和统计相关的指标计算（比如 AUC）。

当时公司也有维护其他的分析型引擎，比如 Druid 和 ES。ES 不适合大批量数据的查询，Druid 则不满足明细数据查询的需求。而 ClickHouse 则刚好适合这个场景。

对于明细数据这个需求：ClickHouse > Druid。
对于维度、指标多的问题，可能经常变动，我们可以用 Map 列的功能，很方便支持动态变更的维度和指标。
按实验 ID 过滤的需求，则可以用 Bloom filter 索引。
AUC 之前则已经实现过。

这些需求我们当时刚好都能满足。

方案设计和比较

常规方案:

比较常规的思路，是用 Flink 消费 Kafka，然后通过 JDBC 写入 ClickHouse。

优点: 各个组件职责划分清楚、潜在扩展性强

缺点: 需要额外资源、写入频次不好控制、难以处理节点故障、维护成本较高

关键是后面两点：由于缺少事务的支持，实时导入数据时难以处理节点故障；ClickHouse 组技术栈以 C++为主，维护 Flink 潜在的成本比较高。

Kafka Engine 方案：

第二个方案，则是使用 ClickHouse 内置的 Kafka Engine。我们可以在 ClickHouse 服务内部建一张引擎类型为 Kafka 的表，该表会内置一个消费线程，它会直接请求 Kafka 服务，直接将 Kafka partition 的数据拉过来，然后解析并完成数据构建。对于一个 ClickHouse 集群而言，可以在每个节点上都建一张 Kafka 表，在每个节点内部启动一个消费者，这些消费者会分配到若干个 Kafka Partition，然后将数据直接消费到对应。

这样的架构相对于使用了 Flink 的方案来说更简单一些，由于少了一次数据传输，整体而言开销会相对小一些，对我们来说也算是补齐了 ClickHouse 的一部分功能（比如 Druid 也支持直接消费 Kafka topic）缺点就是未来可扩展性会更差一些，也略微增加了引擎维护负担。

Kafka engine 原理

这里简单介绍一下如何使用 kafka 引擎，为了能让 ClickHouse 消费 Kafka 数据，我们需要三张表：首先需要一张存数据的表也就是 MergeTree；然后需要一张 Kafka 表，它负责描述 Topic、消费数据和解析数据；最后需要一个物化视图去把两张表关联起来，它也描述了数据的流向，某些时候我们可以里面内置一个 SELECT 语句去完成一些 ETL 的工作。只有当三张表凑齐的时候我们才会真正启动一个消费任务。

这是一个简单的例子：最后呈现的效果，就是通过表和 SQL 的形式，描述了一个 kafka -> ClickHouse 的任务。

最终效果

由于外部写入并不可控、技术栈上的原因，我们最终采用了 Kafka Engine 的方案，也就是 ClickHouse 内置消费者去消费 Kafka。整体的架构如图：

数据由推荐系统直接产生，写入 Kafka。这里推荐系统做了相应配合，修改 Kafka Topic 的消息格式适配 ClickHouse 表的 schema。
敏捷 BI 平台也适配了一下实时的场景，可以支持交互式的查询分析。
如果实时数据有问题，也可以从 Hive 把数据导入至 ClickHouse 中，不过这种情况不多。除此之外，业务方还会将 1%抽样的离线数据导入过来做一些简单验证，1%抽样的数据一般会保存更久的时间。

我们在支持推荐系统的实时数据时遇到过不少问题，其中最大的问题随着推荐系统产生的数据量越来越大，单个节点的消费能力也要求越来越大：

改进一：异步构建索引

第一做的改进是将辅助索引的构建异步化了：在社区实现中，构建一个 Part 分为三步：（1）解析输入数据生成内存中数据结构的 Block；（2）然后切分 Block，并按照表的 schema 构建 columns 数据文件；（3）最后扫描根据 skip index schema 去构建 skip index 文件。三个步骤完成之后才会算 Part 文件构建完毕。

目前字节内部的 ClickHouse 并没有使用社区版本的 skip index，不过也有类似的辅助索引（e.g. Bloom Filter Index, Bitmap Index）。构建 part 的前两步和社区一致，我们构建完 columns 数据之后用户即可正常查询，不过此时的 part 不能启用索引。此时，再将刚构建好数据的 part 放入到一个异步索引构建队列中，由后台线程构建索引文件。这个改进虽然整体的性能开销没有变化，但是由于隐藏了索引构建的时间开销，整体的写入吞吐量大概能提升 20%

改进二：支持多线程消费

第二个改进是在 Kafka 表内部支持了多线程的消费：

目前实现的 Kafka 表，内部默认只会有一个消费者，这样会比较浪费资源并且性能达不到性能要求。一开始我们可以通过增大 消费者 的个数来增大消费能力，但社区的实现一开始是由一个线程去管理多个的消费者，多个的消费者各自解析输入数据并生成的 Input Stream 之后，会由一个 Union Stream 将多个 Input Stream 组合起来。这里的 Union Stream 会有潜在的性能瓶颈，多个消费者消费到的数据最后仅能由一个输出线程完成数据构建，所以这里没能完全利用上多线程和磁盘的潜力。

一开始的解决方法，是建了多张 Kafka Table 和 Materialized View 写入同一张表，这样就有点近似于多个 INSERT Query 写入了同一个 MergeTree 表。当然这样运维起来会比较麻烦，最后我们决定通过改造 Kafka Engine 在其内部支持多个消费线程，简单来说就是每一个线程它持有一个消费者，然后每一个消费者负责各自的数据解析、数据写入，这样的话就相当于一张表内部同时执行多个的 INSERT Query，最后的性能也接近于线性的提升。

改进三：增强容错处理

对于一个配置了主备节点的集群，我们一般来说只会写入一个主备其中一个节点。

为什么呢？因为一旦节点故障，会带来一系列不好处理的问题。（1）首先当出现故障节点的时候，一般会替换一个新的节点上来，新替换的节点为了恢复数据，同步会占用非常大的网络和 磁盘 IO ，这种情况，如果原来主备有两个消费者就剩一个，此时消费性能会下降很大（超过一倍），这对于我们来说是不太能接受的。（2）早先 ClickHouse Kafka engine 对 Kafka partition 的动态分配支持不算好，很有可能触发重复消费，同时也无法支持数据分片。因此我们默认使用静态分配，而静态分配不太方便主备节点同时消费。（3）最重要的一点，ClickHouse 通过分布式表查询 ReplicatedMergeTree 时，会基于 log delay 来计算 Query 到底要路由到哪个节点。一旦在主备同时摄入数据的情况下替换了某个节点，往往会导致查询结果不准。

这里简单解释一下查询不准的场景。一开始我们有两副本，Replica #1 某时刻出现故障，于是替换了一个新的节点上来，新节点会开始同步数据，白框部分是已经同步过的，虚线黄框是正在恢复的数据，新写入的白色框部分就是新写入的数据。如果此时两个机器的数据同步压力比较大或查询压力比较大，就会出现 Replica #1 新写入的数据没有及时同步到 Replica #2 ，也就是这个绿框部分，大量历史数据也没有及时同步到对应的黄框部分，这个情况下两个副本都是缺少数据的。因此无论是查 Replica #1 还是 Replica #2 得到的数据都是不准的。

对于替换节点导致查询不准问题，我们先尝试解决只有一个节点消费的问题。为了避免两个节点消费这个数据，改进版的 Kafka engine 参考了 ReplicatedMergeTree 基于 ZooKeeper 的选主逻辑。对于每一对副本的一对消费者,（如上图 A1 A2)，它们会尝试在 ZooKeeper 上完成选主逻辑，只有选举称为主节点的消费者才能消费，另一个节点则会处于一个待机状态。一旦 Replica #1 宕机,（如上图 B1 B2 ），B1 已经宕机连不上 ZooKeeper 了，那 B2 会执行选主逻辑拿到 Leader 的角色，从而接替 B1 去消费数据。

当有了前面的单节点消费机制，就可以解决查询的问题了。假设 Replica #1 是一个刚换上来的节点，它需要同步黄框部分的数据，这时候消费者会与 ReplicatedMergeTree 做一个联动，它会检测其对应的 ReplicatedMergeTree 表数据是否完整，如果数据不完整则代表不能正常服务，此时消费者会主动出让 Leader，让副本节点上的消费者也就是 Replica #2 上的 C2 去消费数据。

也就是说，我们新写入的数据并不会写入到缺少数据的节点，对于查询而言，由于查询路由机制的原因也不会把 Query 路由到缺少数据的节点上，所以一直能查询到最新的数据。这个机制设计其实和分布式表的查询写入是类似的，但由于分布表性能和稳定原因不好在线上使用，所以我们用这个方式解决了数据完整性的问题。

小结一下上面说的主备只有一个节点消费的问题

配置两副本情况下的 Kafka engine，主备仅有一个节点消费，另一个节点待机。

如果有故障节点，则自动切换到正常节点消费；
如果有新替换的节点无法正常服务，也切换到另一个节点；
如果不同机房，则由离 Kafka 更近的节点消费，减少带宽消耗；
否则，由类似 ReplicatedMergeTree 的 ZooKeeper Leader 决定。

典型案例-广告投放实时数据

业务背景

第二个典型案例是关于广告的投放数据，一般是运营同学需要查看广告投放的实时效果。由于业务的特点，当天产生的数据往往会涉及到多天的数据。这套系统原来基于 Druid + Superset 实现的，Druid 在这个场景会有一些难点：

难点一：产生的实时数据由于涉及到较多的时间分区，对于 Druid 来说可能会产生很多 segment，如果写入今天之前的数据它需要执行一些 MR 的任务去把数据合并在一起，然后才能查历史的数据，这个情况下可能会导致今天之前的数据查询并不及时。

难点二：业务数据的维度也非常多，这种场景下使用 Druid 预聚合的效率并不高。

对比 Druid 和 ClickHouse 的特点和性能后，我们决定将该系统迁移到 ClickHouse + 自研敏捷 BI。最后由于维度比较多，并没有采用预聚合的方式，而是直接消费明细数据。

因为业务产生的数据由 (1) 大量的当天数据和 (2) 少量的历史数据组成。历史数据一般涉及在 3 个月内，3 个月外的可以过滤掉，但是即便是 3 个月内的数据，在按天分区的情况下，也会因为单批次生成的 parts 太多导致写入性能有一定下降。所以我们一开始是把消费的 block_size 调的非常大，当然这样也有缺点，虽然整个数据吞吐量会变大，但是由于数据落盘之前是没法查到数据的，会导致整体延时更大。

改进一：Buffer Engine 增强

单次写入生成过多 parts 的问题其实也有方案解决。社区提供了 Buffer Engine，可以在内存中缓存新写入的数据，从而缓解 parts 高频生成的问题。不过社区文档也介绍了，Buffer Engine 的缺点是不太能配合 ReplicatedMergeTree 一起 工作。如果数据写入到了一对副本（如上图），那么 Buffer #1 和 Buffer #2 缓存的数据其实是不一样的，两个 Buffer 仅缓存了各自节点上新写入的数据。对于某个查询而言，如果查询路由到 Replica #1，那查询到的数据是 MergeTree 部分的数据加上 Buffer #1，这部分的数据其实是和 Replica #2 的 MergeTree 加上 Buffer2 的数据并不等价，即便 MergeTree 的数据是相同的。

针对社区版 Buffer Table 存在的问题，我们也做了相应改进。

(1) 我们选择将 Kafka/Buffer/MergeTree 三张表结合起来，提供的接口更加易用；

(2) 把 Buffer 内置到 Kafka engine 内部，作为 Kafka engine 的选项可以开启/关闭；

(3) 最重要的是支持了 ReplicatedMergeTree 情况下的查询；

(4) Buffer table 内部类似 pipeline 模式处理多个 Block。

这里解释一下我们如何解决查询一致性的问题。前面提到，目前一对副本仅有一个节点在消费，所以一对副本的两个 Buffer 表，只有一个节点有数据。比如 Consumer #1 在消费时，Buffer #1 就是有缓存数据，而 Buffer #2 则是空的。

对于任何发送到 Replica #1 的查询，数据肯定是完整的；而对于发送到 Replica #2 的查询则会额外构建一个特殊的查询逻辑，从另一个副本的 Buffer #1 读取数据。这样发送到 Replica #2 的查询，获取到数据就是绿框部分也就是 Replica #2 的 MergeTree 再加上 Replica #1 的 Buffer，它的执行效果是等价于发送到 Replica #1 的查询。

改进二：消费稳定性增强

由于业务数据的分区比较分散，某个批次的写入往往生成多个 parts。以上图为例，如果某个批次消费到 6 条数据，假设可以分为 3 个 part（比如涉及到昨天、今天、大前天三天数据），第一条和第四条写入到第一个 part，第二第五条数据写入到第二个 part，这时候服务宕机了，没有及时写入第三第六条数据。

由于 ClickHouse 没有事务的支持，所以重启服务后再消费时，要么会丢失数据 {3, 6}，要么会重复消费 {1, 4, 2, 5}。对于这个问题我们参考了 Druid 的 KIS 方案自己管理 Kafka Offset, 实现单批次消费/写入的原子语义：实现上选择将 Offset 和 Parts 数据绑定在一起，增强了消费的稳定性。

每次消费时，会默认创建一个事务，由事务负责把 Part 数据和 Offset 一同写入磁盘中：如果消费的途中写入 part #1 part #2 失败了，事务回滚的时候会把 Offset 和 part #1 part #2 一并回滚，然后从 Part #1 的位置重新消费并重试提交 offset 1-3。

实践 &经验

接入平台

很早的时候为了推广 ClickHouse 消费 Kafka 的方案，我们做了一个接入平台，这个平台可以创建、审核、管理 Kafka -> ClickHouse 的消费任务。前面介绍到了创建一个消费任务需要三张表，直接建表的学习成本比较高，所以这个平台提供这样一个简单易用的页面来完成接入任务。

这里解释一下框起来的两个部分：

首先是查询粒度，如果大家有听第一场分享就大概知道，我们目前对物化视图做了一些改进，假如查询粒度选了 5 分钟或 10 分钟，那消费数据时数据会像 Druid 一样提前对数据预聚合。而查询的时候也会做一些查询改写，用来达到类似 Druid 的效果，目的是为了覆盖公司内部一些用 Druid 的场景。

为简化用户的使用成本，用户也不用挨个填写 Table 的 Schema，而是从 Kafka 的数据里直接推断出 schema。

推断的 schema 是这个样子，用户可以填写一些简单的表达式来做一些聚合 & 转换的工作。

我们也在其他的平台 集成了 Kafka -> ClickHouse 的功能：比如说敏捷 BI 平台，它可以直接支持将 Kafka Topic 作为数据集。

诊断能力增强

前面提到，Kafka engine 方案有一个弊端是增加了引擎端的运维负担。在推广过程中，运维压力也越来越重，我们开发了相应的工具来辅助诊断和运维。我们参考社区 system 表实现了用来辅助诊断 Kafka 消费的表，比如 system.kafka_log 和 system.kafka_tables。

system.kafka_log 表：每当有消费的事件发生的时，会写入一条诊断日志，比如从 Kafka partition 拉取数据对应了 POLL 事件；而写入数据则会对应 WRITE 时间；一旦有异常产生，也会产生专门记录异常的 EXCEPTION 事件。按时间和各类事件聚合之后，可以统计一段时间内的消费量和写入量，也可以通过异常信息去定位和诊断线上问题。

<一个使用案例>

<前面 Query 的查询结果>

如上图，我们可以通过 exception 得知业务方的数据和表的 schema 匹配不上，因而触发了解析异常。同时，也可以到错误消息的 topic、partition、offset。

当然业务方一般不会直接用到这个功能，而是通过运维平台查询消费是否有异常。

这个是 Kafka Tables，这里提供了关于 Kafka engine 的基本信息。最常用的是 assigned partition，可以快速定位某个具体节点消费了哪些 Kafka partition。

SQL 增强

我们也扩展了一些常用的 query，比如 SYSTEM START/STOP 和 RESTART，可以比较快速地关闭、重启消费；比如 ALTER Query 的扩展支持，在不用重建表的情况下变更表的 schema。

未来展望和计划

小结

简单来说在字节内部的应用场景主要分为四类：AB 实验、业务实时数据、服务的后台日志数据、机器的监控数据。Kafka Engine 的改进主要以稳定性改进为主，同时也做了部分性能上的改进。为了方便业务接入，我们也提供了配套的平台和接口，除了自己运维的平台，也和字节内部其他服务做了集成。运维层面则增加了 system table 和 system query 等一系列工具来辅助诊断，简化操作。

目前字节内部仍然没有 MySQL 同步到 ClickHouse 的场景。之前虽然有开发一个简单的方案，但是因为写入和消费不够稳定，并没有在线上使用。因为 MySQL 同步 ClickHouse 时一旦数据出错，那么 ClickHouse 很难再和 MySQL 保持一致了，需要一些额外手段去修复。我们目前有重新设计一套更完整的方案，未来能够支持 TP 数据库的同步，以及支持直接 UPDATE/DELETE Query 直接更新 ClickHouse 的数据。