Druid是Apache 下开源的一款存储与计算一体的olap查询引擎，spark则是纯计算引擎。Druid的数据存储在历史节点，通过broker节点查询，整体的查询流程是两阶段的聚合。数据分布在多个历史节点，查询时，第一阶段在各个历史节点并行计算，第二阶段，多个历史节点的数据汇聚到broker节点，做最后的聚合结算。架构上，broker存在单点瓶颈的风险。通常的意义的聚合，例如sum，max，min，count，broker查询性能足够应对。但是，复杂的聚合，例如去重计算：datasketch，hypeunique，bitmap精确去重，分位数计算，broker显得捉襟见肘，特别吃力。
这类计算不是简单的类型的加减乘除(max,min,sum)，而是复杂数据容器之间的计算，占用堆内存特别多，CPU消耗也很多。顺便说一句，我们的分析场景，max，min，sum，avg，count 毫无地位，disctinct count才是主角，恰恰去重计算业界根本没有特别好的解决方案。唯一的一个是美图的基于分布式bitmap的去重计算：https://www.infoq.cn/article/UEyPTRJcztM7FQGebtfT
单个broker根本不足以应付。解决思路自然是增加broker节点，增加聚合的并行度。
1）数据分布在60个历史节点上，我们可以引入10个broker节点，每个broker节点负责聚合6个历史节点聚合的中间数据。这样第二阶段有10个broker节点在聚合。最后再引入一个broker，将10个broker的聚合结果最后一次聚合。
image.png
2）1）的不足之处在与，第二阶段的聚合的并发还是不够多。直接一步到位，引入分布式计算引擎：spark或者presto，历史节点的计算的结果在spark中构建临时表，通过全体数据shuffle的方式，进行聚合。这里以spark为例。
假设数据分布在60个历史节点，spark启动60个executor，每个executor分别拉取一个历史节点的数据，这样RDD的分区数量是：60。通过自定义UDAF函数，对数据通过shuffle进行分组聚合。最后的结果拉倒driver。这里的最后结果已经最终的结果，可以直接传输到driver，返回查询结果。
image.png
1）自定义RDD。实现compute，getPartition等接口。
这里定义RDD的分区数量与传入的参数历史节点的数量相等。
定义next接口。这里强调下spark和druid是一脉相承。druid中，历史节点返回聚合数据的方式就是通过迭代器的方式流式返回，所以，RDD的next接口中，可以自然使用druid中迭代器接口：JsonParserIterator。直接将DirectDruidClient中JsonParserIterator封装一下，放到RDD的next中，特别合适。复杂聚合类型的数据类型，历史节点中返回是byte[],在spark需要转换为binaryType。
2）自定义数据源。实现relationprovider接口，buildscan等。
3）自定义复杂类型的聚合函数UDAF。
可以参考这个https://www.shangmayuan.com/a/f5496e01261d4b499d9cb938.html
本质上就是将byte[] 反序列为RoaringBitmap类型，进行与或操作，然后序列化，开始下一次聚合。
网友评论