从需求场景下出发实操Clickhouse

Flink实战剖析

发布于 2022-04-18 05:38:25

6230

背景

本着以实时数仓为目标调研了几款OLAP引擎，像Clickhouse、Kylin、Druid等，在粗略了解其架构后，并且在接受各个大厂Clickhouse实践、高性能测试报告、最近业界发展势头凶猛的熏陶与PUA情况下，不得已选择了Clickhouse，当然自己也做过一些测试，本篇将介绍clickhouse的一些原理、实践方案(可能还未实现、可能并不是最佳)与遇到的一些问题，总之只是希望能够为您接下来选择clickhouse 或者解决一些问题提供一个参考的思路，仅此而已。

多维聚合

不得不说Clickhouse的这个操作确实是很快，应该也是吸引大家的地方。

首先是这种Shared nothing架构，每个节点都可以拥有自己的存储、cpu、内存，资源不共享，每一个节点都可以单独使用，使得其横向扩展能力很强，自然而然就增强了其计算能力；

列式存储方式，列式裁剪，每一列都生成单独的存储文件，不管是在数据压缩还是查询情况下都会有高性能的表现，因为只查询自己关心的列，可以快速过滤。补充一下只有在wide 的存储格式下才会每一列使用单独的存储文件，另外一种称之为compact，所有的列在同一份文件，这两种格式的控制是由写入数据量大小解决的，数据量小就是compact，那么就可以提高写入的性能了；

向量化执行，通过SIMD指令一次能够操作多条数据，大大的提高计算性能，基本上在所有的OLAP里面都有这个功能，该指令需要底层硬件层面的支持，果然硬件的提升使得性能的提升来得更直接、更有效；

并行执行，一个是多节点的并行计算，另一个是单节点的多线程并行计算，可以小到每一个数据块data-part多线程执行，充分的损耗你的CPU，来加速它的查询速度；

分区裁剪与索引，分区裁剪就很熟悉，数据按照指定分区方式组织，分区条件可过滤大部分数据出来，再说说索引，我且称之为块索引吧，按照主键排序的方式构建索引，我们比较熟悉的是B+树这类索引，为啥不使用B+树？我的理解是B+树建索引的话，如果你需要多个索引字段，那么你可以构建联合索引，那么查询的时候需要满足最左原则，但是你单个构建索引的话，文件比较多，会消耗存储，所以我想clickhouse这种方式就避开了这个问题，另外你还可在clickhouse主键索引上构建一些跳数索引进一步加快搜索过滤；

事务支持

OLAP这种分析存储引擎是一次写入多次读取，所以一般不需要支持事务的，为啥我们还一直在提这个，看看这两个场景(其他的读写并发就不扯了)：

实时写入，在实时数仓的整个流程中，通过Flink将数据写入到Clickhouse，要使Flink保证端到端的一致性，上游的数据重放机制、中间状态机制以及下游输出幂等支持，上游重放由对接kafka保存offset来实现，中间状态Flink天然支持，下游就比较难以实现，你可以做幂等或者事务支持，可以看看kafka/hdfs-connector这两个部分的实现，但是clickhouse 是不支持事务，你就别想使用两阶段提交了，在异常情况下数据是存在重复写入的情况的(先不要考虑ReplacingMergeTree了)。大家不都说实时数据不准确吗，那就是不准确吧，直接上Lambda架构，做一份离线覆盖吧；
离线导入，离线导入有很多现成的工具可以使用datax、waterdrop，或者你自己写一个spark任务，不管怎么样一般失败了任务就会重跑，通常也是按照分区导入的，在导入前会做分区数据删除，看起来是没有什么问题，但是对于spark这种，task失败了中间会有重试机制，重试机制又没办法关闭最小值就是3，所以你得尽可能的让你的任务稳定，让其中间不必重试，另外得加一个校验机制，比对clickhouse数据与源头数据一致性。另外一种没有校验的方式，clickhouse 本身支持hdfs 类型的表engine，那我们可以通过spark任务生成hdfs数据(这里的重试数据不会重复)，然后通过hdfs表engine的方式导入到mergeTree表engine中。最后关于离线导入的姿势建议大家参考腾讯的clickhouse实践案例，离线大批量的导入对clickhouse集群会有比较的的io消耗，并且伴随着数据文件的增多，后台的merge消耗也会比较高，对cpu会造成一定压力，势必会影响线上业务，通过构建临时节点方式生成对应表结构的数据文件然后进行分区attach方式。

关于更新

要更新，真是哪壶不开提哪壶，战斗民族都说了我们最开始就是用来做流量分析哪里还需要更新操作，别扯淡了，你可能会说我们不仅要做应用日志这种insert类型的，但是我们还有一些像mysql binlog这样的数据呢，你就不支持了？老毛子喝着伏特加就来了一个alter，你们就这样用吧，但是它是一个异步的过程，会比较消耗io。关于更新我们尝试了两种方案：

Alter 方式，接收kafka数据每一条数据都进行一次delete+insert的编码，批量写入方式，写入是没问题，但是后台观察到io蹭蹭的往上涨，很恐怖，直接放弃了。但是也有可能我们的场景比较特殊，或者更新比较少可以尝试一下；
ReplacingMergeTree方式，这种表引擎会按照Order by 指定的字段在后台进行合并操作，但是你在查询的时候还是需要带上去重的方式，我们使用的方式是在其之上创建view, view里面以原表里面Order by 字段作为Group by字段，其他的字段使用argMax方式取值。但是使用这种方式需要保证相同的order by字段在同一个节点上，不然ReplacingMergeTree没法进行数据合并，会导致无用数据越来越多。大多数情况我们都是写本地表，按照这种方式，我们在客户端需要自己做路由，因为连接LB的方式会比较随机，同样的这种比较固定的hash方式，在后期遇到节点扩容需要进行数据的rehash操作，运维成本会相应的增加。

数据模型

这里所说的数据模型是指数仓里面的模型，通常会使用维度建模方式，用事实表与维表整一个大宽表这种方式，因为我们在往clickhouse里面写数据的时候就需要这种考虑，通常分为这两种方式：

大宽表模型，也就是在写入数据到clickhouse之前将数据join起来，clickhouse本身在这种单表分析下具有比较强的优势，但是宽表从另外一个层面来说也是会损失一定的数据正确性，因为维表的变更，会导致当前相同事实数据处于不同的维度，当然这个是在实时的场景下，使用离线覆盖就好啦；
星型模型，将事实表与维表直接导入到clickhouse中，在查询的时候做关联，也就是说要join了,虽然说提供了global join的方式优化，但是join 毕竟需要数据的跨网络分发，数据量一大，网络io就会成为瓶颈，我们可以使用colocate join这种方式，将两张需要join的表的key写入到相同的节点，然后以本地表的方式进行查询即可，就不会存在网络的数据分发，但是这种方式同样在后期遇到节点扩容需要进行数据的rehash操作，运维成本会相应的增加。

并发性与灵活性

并发性一直都是大家对Clickhouse感到尴尬的地方，因为Clickhouse会拼尽全力去执行一个Sql查询，让其能够更快的得到查询结果，但是我们总想着查的快可以，能不能让我的查询并发更多一些呢，我们在做这方面的测试，发现了并发100个就很不错了，当然是跟业务相关的查询。另外也可以做一些其他的查询优化，比喻说做物化视图，这种以空间换时间的方式，或者是说在查询层前面做一个缓存，将查询的Sql语句与结果缓存起来，下次相同的查询直接从缓存里面获取即可；

灵活性确实很强，分布式你自己搞、副本你自己搞、数据重分布你自己搞、多节点的配置你自己搞等等，总之我有这个功能，用得怎么样看你自己了，你搞一个比较强大的运维支持就没事了，总之好不好还是得看你们自己。