ClickHouse在字节跳动推荐和广告业务部门的最佳实践

爱被打了一巴掌 2022-11-04 14:25 227阅读 0赞

此次分享分为三部分内容，第一部分通过讲解推荐和广告业务的两个典型案例，穿插介绍字节内部相应的改进。第二部分会介绍典型案例中未覆盖到的改进和经验。第三部分会提出目前的不足和未来的改进计划。

# **一、早期实践** #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70][]

### 1、外部事务 ###

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

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

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 1][]

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

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

**1）直接写入的风险**

用户写入 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。

# **二、典型案例-推荐系统** #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 2][]

**1、业务需求**

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

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

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 3][]

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

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

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 4][]

**2、方案设计和比较**

**1）常规方案**

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

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

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

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 5][]

**2）Kafka Engine 方案**

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

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 6][]

**3）Kafka engine 原理**

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 7][]

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 8][]

**3、最终效果**

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

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 9][]

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

**改进一：异步构建索引**

第一做的改进是将辅助索引的构建异步化了：在社区实现中，构建一个 Part 分为三步：

*  解析输入数据生成内存中数据结构的 Block；
 *  然后切分 Block，并按照表的  schema 构建 columns 数据文件；
 *  最后扫描根据 skip index schema 去构建 skip index 文件。三个步骤完成之后才会算 Part 文件构建完毕。

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 10][]

**改进二：支持多线程消费**

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

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 11][]

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 12][]

**改进三：增强容错处理**

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

为什么呢？因为一旦节点故障，会带来一系列不好处理的问题。

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 13][]

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 14][]

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 15][]

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

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

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

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

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

# **三、典型案例-广告投放实时数据** #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 16][]

**1、业务背景**

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

*  产生的实时数据由于涉及到较多的时间分区，对于 Druid 来说可能会产生很多 segment，如果写入今天之前的数据它需要执行一些MR的任务去把数据合并在一起，然后才能查历史的数据，这个情况下可能会导致今天之前的数据查询并不及时；
 *  业务数据的维度也非常多，这种场景下使用 Druid 预聚合的效率并不高。

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

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 17][]

**改进一：Buffer Engine 增强**

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 18][]

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

*  我们选择将 Kafka/Buffer/MergeTree 三张表结合起来，提供的接口更加易用；
 *  把 Buffer 内置到 Kafka engine 内部， 作为 Kafka engine 的选项可以开启/关闭；
 *  最重要的是支持了 ReplicatedMergeTree 情况下的查询；
 *   Buffer table 内部类似 pipeline 模式处理多个 Block。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 19][]

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

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 20][]

**改进二：消费稳定性增强**

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

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 21][]

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

# **四、实践&经验** #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 22][]

**1、接入平台**

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

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

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

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 23][]

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 24][]

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 25][]

**2、诊断能力增强**

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 26][]

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 27][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 28][]

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 29][]

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 30][]

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 31][]

**3、SQL 增强**

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

# **五、未来展望和计划** #

**1、小结**

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 32][]

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 33][]

**2、未来展望**

未来我们计划持续投入人力来完善 ClickHouse 写入的功能。

*  第一步是实现在 ClickHouse 上分布式事务，以此解决 Kafka engine 消费以及 INSERT Query 不稳定的问题；
 *  之后会尝试实现读写分离，将消费数据的节点与查询节点隔离；再基于读写分离做消费节点的动态伸缩；
 *  分开的消费/写入节点做为专门的写入层，后续会引入 WAL 和 Buffer 解决高频写入的问题。如果有必要的话，也会在写入层实现类似分布式表分发数据的功能；
 *  一旦上面的功能实现成熟，会考虑重新开放业务直接使用 INSERT Query 写入数据。

作者丨查询分析团队

来源丨字节跳动技术团队（ID：toutiaotechblog）

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70]: /images/20221023/3833b588278d467a8973305a1cb8991d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 1]: /images/20221023/978796102a6647c6a8690fb791f66b6d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 2]: /images/20221023/82baf48eaad8425b9c6bfeac48e8eb12.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 3]: /images/20221023/5d3f5398ed1d436fa92a342ad37fff5d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 4]: /images/20221023/b4d4bbfd95a343daaad6a527a95edc37.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 5]: /images/20221023/5d7a8556b833445b9f9b98cb38655cca.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 6]: /images/20221023/5f8d59dfe77647c09e45a241252e595a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 7]: /images/20221023/fe93c52b00814d5bb050f87efc45a2bd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 8]: /images/20221023/9b1e72c41c08479681a4dcf4cbadbd81.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 9]: /images/20221023/e5a72dcb605c4904a2a68483c463642b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 10]: /images/20221023/b7b89e0a2f7144da8d3de2208e905b7f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 11]: /images/20221023/70aa92f26f2e46d083f795c9231ae91e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 12]: /images/20221023/521f202a796b45af9aaebcd37de8ee68.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 13]: /images/20221023/91da9a4fde8e44938d5bb80d4a18caa9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 14]: /images/20221023/550e844f76124da6acb70a08206d4a4b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 15]: /images/20221023/aa064a37c14a44f8b026fb542d8f6263.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 16]: /images/20221023/98f2891174134406b53c40752385cbcf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 17]: /images/20221023/ed07afef56484825b9a10ceeeaa8faae.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 18]: /images/20221023/647c95699a2846f9b5229fd357424de1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 19]: /images/20221023/ae82bbae204746819b2638ac35ed3664.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 20]: /images/20221023/a85827ba6f4a4314b743f01ac9df9878.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 21]: /images/20221023/b71fc2bdaafc4bc88c31e7e59dd71e1b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 22]: /images/20221023/88cdc1d1c9c04adda645d670756f3e9f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 23]: /images/20221023/7dd23589eaf641b285eb85a6b57319e6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 24]: /images/20221023/89bfbdc1ab2049c8973a407e989d00f9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 25]: /images/20221023/a765a195ea704478b758594b9737b0b8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 26]: /images/20221023/320f8a57fe80479fa4c5e6e71a316af6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 27]: /images/20221023/a3b9282d33e547bfaee2953181ae41c5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 28]: /images/20221023/a3cd97609aa4443fae33b331ff2e037b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 29]: /images/20221023/84d04d5a0d9e4f1fab7b6f375345d837.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 30]: /images/20221023/ae4019218c37417785acfecb02077e10.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 31]: /images/20221023/60d03fb7dc3742749db8b2af0a415406.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 32]: /images/20221023/98feb35acaf9416892234dc1bfc7a5dc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2x1b3Fpbmdsb25nODUwMTAy_size_16_color_FFFFFF_t_70 33]: /images/20221023/a500651110ab4446a4daff3159ea90da.png