大数据架构系列：Apache Kylin 4.0

jhonye • 2023-01-02 • 云技术社区 • 294 阅读

背景

身处于大数据时代，即使我们使用的大规模并发对数据进行查询，由于数据量的原因，用户想快速的对数据进行分析还是较为困难的；预计算是其中一个比较直观的解决方案，提前将数据算好，需要的时候直接拿出来，看上去是非常美好的，但是预计算是需要成本的，由于分析场景的复杂，预计算的结果被复用的概率可能没那么高，但是这一步还是需要有人进行探索和实践。本文主要描述了Apache Kylin 4.0.1的原理来帮助大家打开思路。

架构

整体架构

上图源自官网，我们可以看到几个核心的模块：构建引擎（Build Engine）、查询引擎（Query Engine）、Cube数据（OLAP Cubes）、元数据（Metadata）、Web服务（REST Server），其中Routing可以归到查询引擎。用户可以通过JDBC/ODBC或者REST API的方式进行访问。

Cube数据

Cube 的直观图

如图，（A,B,C,D）为4个维度按照GROUP BY的方式进行分层组合，最终4个维度的组合会存储到一个目录包含14个文件；既然是GROUP BY肯定会数据中会包含聚合函数，只会在往下层继续聚合。

这块在Cube Planner的逻辑中可以进行优化，控制计算和存储成本。

构建引擎

构建引擎的作用是将来Hive、Kafka等的数据转化为Cube数据，构建时会直接拿N维组合的结果去计算N-1维组合的数据；构建速度上一般不太强求，尽可能保证构建成功，一般选择通用计算引擎MR、Spark。

查询引擎

在Cube数据生成好后，我们就可以基于该数据进行查询；查询引擎会将用户的sql 进行切分，不同的子查询可以命中不同的Cube，以此来快速响应用户的SQL请求。

当前版本支持在没命中Cube时，直接查询原始数据；不支持将一条SQL拆成部分查询Cube数据，部分查询原始数据。

Cube构建

Cube构建可以抽象为Source、Tranfrom、Sink的过程，需要读取有Schema的原始数据，通过Cube的构建规则预处理，最后按照Cube的组织结构将数据写入到存储中；目前Cube的存储格式使用Parquet，Cube数据的Schema来源于原始数据。

Kylin构建的Cube数据不会随着用户原始数据的更新而自动进行增量更新，需要用户主动进行维护。会存在原始数据与通过Cube计算的结果不一致，可以理解当前的Cube数据只是原始数据某一个时刻的镜像。

用户在创建好Kylin的Model、Cube后，就可以对原始数据进行构建，一般情况下会配置分区日期列（Partition Date Column）进行增量构建，每次增量构建选取的时间范围为一个Segment，也可以不配置分区日期列则进行全量构建。

全量构建

全量构建为增量构建的一个特例，即构建全部分区的数据，只能通过刷新构建的方式来更新数据。

增量构建

用户指定好一个时间范围后，构建时则会在 WHERE 条件指定该范围的数据进行预计算，计算完成后将数据存储到指定目录，最后commit元数据，生成一个新的Segment，表示该范围的Cube数据可以使用。接着就是追加一个个Segment的数据即可。

Segment 元数据

Q&A

Q1. 每次构建的Segment数据都是独立的，那么在查询时如何将数据汇总呢？

A：预计算的规则是受限的，聚合函数只支持COUNT、SUM、MIN、MAX、COUNT DISTINCT、近似算法。是可以直接合并不同分区的数据得到正确结果，如果不正确就不支持。

Q2. 每天一个分区，最后会不会导致文件过多，元数据也会爆炸，导致查询变慢。

A：Kylin内部会有促发条件（用户可以配置），可以按周、月、年在后台进行合并成一个新的Segment，即后面的刷新构建。

Q3: COUNT DISTINCT 精确去重为什么可以多个Segment合并

A：全局字典 + Roaring Bitmap

Q4：流式数据如何构建？

A：新增一个个小的Segment，然后定时、按Segment个数合并，类LSM结构Compaction。

刷新构建

用户在出现Cube数据与原始数据不一致时，可以对历史Cube进行刷新，在构建完成后替换原来的Segment的元数据，并不会立马删除老的Segment数据，防止有查询已经命中老的Segment导致查询异常，会在后续的垃圾清理机制中，定期清理掉无效数据。

构建详细流程

一般情况下，用户会基于维度建模的方法论创建Cube，一张事实表和多张维度表，所以Kylin的第一步是需要进行打平表的，即通过JOIN生成一张大宽表。大宽表包含的列只有用户选取用来构建Cube的列，Measures中使用到的列也算在里面。
例如有（A,B,C,D）4列，那么会先构建基于（A,B,C,D）GROUP BY的结果，称为Base Cuboid。
接着会基于Base Cuboid 去计算 GROUP BY ABC、ABD、ACD、BCD 4个Cuboid数据，即上卷操作。
依次计算到只有一个维度的情况。
最终将数据按照Cube格式存储。

物理存储格式：

Cube数据存储的结构

逻辑格式：

Cube数据的维度组合关系

Cube构建优化

在生成Cube数据的过程中有许多地方可以进行优化，大部分业务场景下我们不仅需要极速的查询体验，也要尽量压低计算和存储成本。

优化Cube

1. 可以通过配置必要维度、层级维度、联合维度对Cube的维度组合关系进行主动减枝。

2. 开启Cube Planner，并指定优化目标，后台可以根据成本/效用的情况进行自动减枝。

3. 对Spark/MR的构建任务进行调优，多表Join可以加Hint，配置broadcast等。

Cube查询

在我们费力将Cube数据构建好之后，我们就可以使用Sql进行查询；当然不需要直接去查询Cube数据，我们可以还是写查询原始表数据的Sql，Kylin会将Sql改写优化命中Cube的部分Cuboid数据进行回答。速度自然是大大提升。

举例：

Cube - 原始表：Table，维度：A,B,C,D， Messure：SUM(E), COUNT(1)

用户SQL：SELECT A,B,C,SUM(E) FROM Table GROUP BY A,B,C;

结果：此时会直接命中 A,B,C 组合的Cuboid来回答用户的查询；上面的用户SQL可以是一个复杂SQL里面的一个子查询，子查询不能嵌套子查询。

查询详细流程

Cube 查询流程

用户使用JDBC/ODBC或者REST API的方式发送一条查询SQL到Kylin的REST Server。
REST Server做一些简单的权限检查，判断缓存里是否有结果可以直接回答。
确认需要继续解析SQL，则创建Calcite的connection，元数据使用Kylin数据库中存储的元数据，为用户主动关联导入的。
statement调用CalciteMetaImpl#prepareAndExecute进行SQL的解析、验证、优化；SQL需要符合Kylin的语法，根据VolcanoPlanner里的优化规则，迭代做一些SQL的优化，也为了转成更容易命中Cube的逻辑。
接着核心点是执行到implement(root) -> OLAPToEnumerableConverter#implement 这块逻辑是Kylin自己实现的，将SQL中可以直接扫描数据的子查询转为OlapContext，实现逻辑都在org.apache.kylin.query.relnode包下面。
如果RealizationChooser#selectRealization发现存在OlapContext不能找到对应的Cube数据回答，则抛“No model found for” 异常，路由到直接查询原始数据的分支进行下推查询（pushdown）；如果全部OlapContext都找到对应的Cube进行回答，通过代码生成org.apache.kylin.query.exec.SparkExec.collectToEnumerable(root)相关逻辑用于最后计算数据。
接着调用CalciteResultSet#execute -> org.apache.kylin.query.exec.SparkExec.collectToEnumerable(root)，将OLAPRel 转为 Spark的物理执行计划并执行，具体转化逻辑参考CalciteToSparkPlaner#visit 。
通过Spark的DataSet执行完结果后返回Enumerable<Object>迭代类型数据。
最后QueryService#createResponseFromResultSet封装好结果返回给用户。

Q&A：

Q1: 如果出现多个Cube都可以回答对应的OlapContext的话选取哪个？

A：通过CubeInstance#getCost的值做对比，取最小值。公式：模型维度数*10 + Measures * 1 + INNER JOIN维度表数量*100

Q2: 如何Kill掉一个查询

A：REST Server内部每个查询都是一条独立的线程负责，中间会设置许多可以中断的点。

结语

通过上述的分析，我们发现Kylin4的新架构在设计和实现上确实比较优秀，可以在大量的场景下帮助用户进行透明加速查询，整体逻辑还是比较符合维度建模的理论。另外Kylin是有商业化产品的，4.0的新架构也是从商业产品转化过来，功能差别并不大，在产品化上会做的更好，例如Schema change的自动更新方式、给用户自动推荐模型/Cube等。

同时也存在许多不足之处，例如数据一致性需要用户自己保证，复杂查询场景无法支持，预计算成本较高等问题；但是没有一个架构是完美的，我们要做的是在前人的基础之上去改进，做出更优秀的产品。