Apache Doris 在有赞的初步尝试

作者:郑生俊

有赞OLAP

有赞作为一家商家服务公司，OLAP对有赞非常重要，从商家后台的数据看板，再到最近有赞门罗发布会上的有赞贾维斯，都离不开OLAP。具体到技术栈，有赞的MOLAP采用Apache Kylin（参考文章：有赞出品｜升级 Kylin 4 最强攻略！），ROLAP 采用了ClickHouse（参考文章：ClickHouse 在有赞的实践之路）。

Apache Kylin在有赞内部已经非常稳定了，支撑了大量的高QPS的场景，并且它的存储计算分离架构能够很好地做到弹性伸缩来应对流量的高峰低谷。而ClickHouse作为ROLAP引擎最主要的问题有两块，一是扩缩容麻烦，二是单机的Join性能差。其中扩缩容的高成本是有赞数据团队在ROLAP最为头疼的问题，比如有赞门罗发布会发布了MA的优惠之后，因为ClickHouse扩容的时效性和复杂度，我们需要提前准备较多的硬件为ClickHouse扩容，以应对有可能的业务剧增。尤其是如果ClickHouse之上又有实时写入的场景、做了数据正交分布时，扩容就难上加难了。

试水Apache Doris

为了解决上面的问题，团队也尝试了ClickHouse on Apache Doris，但后来因为一些计划变动没有继续投入。欣慰的是Apache Doris在今年推出了向量化引擎，也在6月15号正式孵化为Apache顶级项目，它的设计上能够解决ClickHouse上述的问题。进而细看了Apache Doris的向量化引擎，基本上借鉴了前辈ClickHouse的做法。既然是站在巨人的肩膀上，我们觉得它应该是能够承载ClickHouse之上的业务。为此我们做了初步的性能测试和与Druid的兼容性测试。

查询性能：

在初步的性能测试过程中，结果还是比较惊喜的，比Druid快很多，也比之前非向量化的版本快很多，有些场景和ClickHouse差不多。查询这块主要分享一下我们最近在2phase aggregate做的优化。

背景是我们测试在merge aggregate的过程中，发现第二阶段的聚合比第一阶段的聚合速度慢很多。首先解释一下merge aggregate，对于分布式聚合查询，数据通常都需要在某个算子预聚合后，再汇聚到下一个算子进行merge聚合。但我们从Doris的执行信息看到第二阶段的聚合比第一阶段的聚合慢不少，而测试场景下，二阶段聚合的数据量比第一阶段的聚合少很多，反而二阶段聚合的耗时还更高了。

首先尝试使用perftools来分析瓶颈点，找到核心的耗时代码：

我们找到上面代码，看了一下二阶段聚合的大致流程，大概有这么几步：

1. 反序列化一阶段的聚合结果，得到StringColumn
2. 将StringColumn，反序列化为对应列的数据类型（double、long、hll 等数据结构）
3. 将对应的数据列（double、long、hll 等数据结构），转换为临时用来聚合的数据类型（AggregateData）
4. 将临时的聚合对象与Hash表中存放的最终结果（也是AggregateData）进行聚合运算
5. 销毁3步骤产生的临时聚合对象

结合上面的perf采样图，我们可以看到大量的CPU消耗在：

1. 将StringColumn反序列化为对应的数据类型，比如long、int、hll 等
2. 将聚合的输入数据（long、int、hll）等转换为聚合需要的AggregateData，这里就有大量临时 对象/内存 的频繁创建和销毁

为了解决上述第一个问题，尝试了在第一阶段聚合结束之后，直接将结果转换为第二个阶段需要的数据类型。这样第二阶段的数据读取，只需要做对应指针类型的强制类型转换即可，而不需要再从StringColumn反序列化获得。

为了解决第二个问题，起初我尝试着分配整个块内存来存放要聚合的临时数据，然后最后释放整块内存，以此减少频繁的内存创建和销毁过程，但收效甚微。这时候我们去看看第一阶段聚合的代码为啥会更快，把原因说的通俗点就是：AggregateData可以直接与Block中的数据（double、long、hll 等数据结构）进行聚合运算，不需要将其转换为聚合的相同数据类型（AggregateData）进行聚合运算即可，这样避免大量 对象/内存 频繁的创建、销毁。

按照上面的思路改完代码之后，我们拿一个测试环境的例子看看性能提升，这是一个涉及到600w+数据读取的查询。查询了多次让磁盘数据进操作系统的cache之后比较RT，查询总耗时优化前810ms，优化后560ms，提升了30%的性能。

优化前：

优化后：

虽然这个case从整体上看RT提升250ms，看着好像可有可无，但这是在降低了资源消耗的情况下达成，往往就意味着系统能够有更高的吞吐。而且数据量大了、查询更复杂之后会有更大的提升，毕竟Aggregate在OLAP中是一个高频操作。

从整体查询RT的视角，RT容易会有波动，但算子内MergeTime的指标统计不会，因为MergeTime指标统计的都是针对一个内存中Block的聚合耗时，排除了很多网络、IO的干扰因素。

优化前：

优化后：

从上图可以看到MergeTime也大大降低了（其中 ExecTime 指标的变化可以忽略，因为走了不同的代码逻辑，是包含在MergeTime的统计之中的）。

好在Doris的代码写的挺合理的，所以这个优化涉及到的代码量不多，有兴趣可以参考代码：https://github.com/apache/doris/pull/10618/files。

兼容性

由于我们第一阶段的目标是将Doris替换Apache Druid，因此我们基于Druid场景做了一些兼容性测试。对于平台型团队而言，一旦上层有较多的业务，要推动底层的技术栈迭代和替换是比较困难的，因为通常业务方也很难抽出时间和我们陪跑做全面的兼容性测试。

因此我们在Druid Broker处理完查询之后，将一个查询的SQL、RT记录数等信息发送到Kafka。然后由一个Kafka消费者消费上Druid的请求，做SQL改写，将Druid SQL转换为Doris的查询语句发往Doris进行流量回放。

‍

一些SQL的基本语法都是相同的，比较大的区别是 builtin 函数。有一些Druid Function 的函数的参数入参含义、个数，都和Apache Doris有较大不同，这导致SQL改写的过程繁琐一些，但这对于平台型的服务团队通常是不得不做的过程。通过流量回放一来可以知道哪些不兼容的SQL语句要做什么样的调整，二来可以通过模拟线上查询情况的过程中确定哪些性能不符合性能预期、哪些查询是有Bug的，比如Druid时序数据查询中经常使用到的time_round_function计算有误（参考代码：https://github.com/apache/doris/pull/9712/files）。确定了上述的各方面的性能和兼容性没问题后，我们才能更高效地协调业务方做一些改造工作。目前我们已经回放了一部分线上查询，整体的业务改动点还好，性能也有较大提升。

后续计划

至于后续的计划，我们的目标是视资源情况推进Doris在有赞落地，尽量将ClickHouse、Druid的技术栈收敛为Apache Doris，解决前面提到的问题，同时也做技术栈的收敛、迭代。当然这还有一些工作要做，包括兼容性测试、性能测试，确保业务上Doris与ClickHouse、Druid有相当的体验，为此我们也在尝试一些手写SIMD优化关键执行代码，希望最终能够借助Apache Doris解决我们ROLAP的痛点问题。