滴滴：Apache Kylin 自助式治理与演进之路

作为开源社区，Apache Kylin 社区的成长离不开用户在代码、案例、文档等方面不断的贡献。在 2019年，Apache Kylin 变得更为稳定，功能也更为丰富，以历史数据分析见长的 Kylin 开始涉足实时数据分析领域，社区力量也在不断壮大。

在 7 月 12 日的 Kylin Data Summit 上，来自滴滴出行的技术专家靳国卫获得“2019 最佳 Apache Kylin 社区贡献个人奖”。靳国卫从 2015 年就开始接触 Kylin，向社区贡献了 Livy 集成、利用 Hive 构建全局字典等功能。

当天下午，靳国卫也在大会的互联网专场上进行了 Kylin 在滴滴的实践分享，下文为现场的精彩分享内容。此次分享将分为上、下两期。本期首先为大家介绍滴滴当前的 Kylin 架构以及滴滴如何进行集群的治理。

大家下午好，我叫靳国卫，来自滴滴出行。上午大佬们预测了技术的前沿领域和发展方向，下午我就从细节讲一下我们在实际应用过程中遇到的问题，如何解决这些问题。

PPT 主要分为四个部分，第一部分讲一下 Kylin 架构在滴滴的实际情况，展示集群数据规模，进而引出一些问题；第二部分讲在滴滴如何进行集群的治理；第三部分结合具体实际场景讲为什么做字典改造，及其改造的过程和收益；最后留一些时间回答下大家的提问。

平台架构、数据展示

此图是 Kylin 数据服务架构。滴滴的数据场景大部分都是自助服务式的。主要过程如下：

1. 用户自己通过可视化平台完成数据分析、报表发布。这个过程动态产生数据模型、ETL 数据生产、Cube 创建与调度等。
2. 建模、资源隔离、智能调度、版本管理模块负责数据的生产。比如，用户自助创建了一张报表，这时候平台需要根据用户的操作过程完成数据建模，完成调度创建，集群选择，多次发布带来的数据版本管理。而用户只需要关心数据来自哪里，数据可视化如何组织。
3. 集群的核心服务是查询，比如，接收到查询请求后，查询中心会结合元数据中心判定本次查询是否命中了 Kylin 的 Cube，如果命中 Cube 则从 Kylin 集群获取数据，否则通过 Adhoc 查询从 Spark 或者 Presto 获取数据。
4. 自动化运维和集群管理模块后面会详细介绍。

目前我们一共部署了 7 个 Kylin 集群，分国际和国内集群，服务基本上涵盖了公司的绝大多数服务。每天构建任务 7000 左右，Cube 数量 5 千以上。

为什么需要做集群治理呢？是因为绝大多数服务都是自助拖拽式完成的，所以主要压力是 Kylin 集群原数据膨胀。下面讲下滴滴是如何治理 Cube 元数据的。

服务治理

1. 基本架构

上文中的架构图中的 Kylin1、Kylin2 都是一个 Kylin 集群。具体到单个集群，分三类角色节点：query、build 和 status。然而 Kylin 本身只有 job 和 query 两种节点类型，后面会详细讲为什么要增加 status 类型的节点。

Kylin 集群的核心是查询能力，查询服务是它的 SLA，要求查询服务稳定可靠，所以 query 节点通过 VIP（Virtual IP）来实现查询的负载均衡。

我们每天有 7 千个任务需要构建，构建节点相应会多一点，build 节点可以设置构建任务并发量。但是因为字典的构建是在 build 节点单机完成的，如果并发量设置太高，会对 build 造成一定压力，所以要根据负载合理设置并发数。

status 节点提供 API 服务与开放平台进行交互，实现集群元数据变更。为什么要有 status 节点，为什么要实现 active/standby 在下文会有详细解释。

2. 集群治理

集群治理从几个方面来说明：

① 为什么会有多集群？

1. 根据实验后发现 Kylin 节点所占用资源其实不大，所以将 Kylin 节点实现为一个 Docker 节点；
2. 当一个集群中 Kylin 的元数据达到一定程度，稳定性会有一些问题，各个业务对服务稳定性要求不同，要求独立部署。

基于这两个方面考虑，就实现了集群、节点动态伸缩功能。

② 既然有多个集群就会有几个问题：

1. 业务和集群如何进行资源绑定；
2. 集群间负载不均衡如何做压力转移。

所以要实现资源的动态转移。

③ 为什么要实现 Status 节点的 HA？

节点的部署对象是 Docker 镜像，里边包含了 Kylin、Spark、HBase、Hadoop、Hive 的 rpm 包，在Docker 启动的时候会随系统启动一个 kylin-agent，agent 主要负责 Kylin 启动、资源管理、存活监控的工作。

节点启动时，会去配置中心读取当前这个节点的配置信息，包括节点类型、属性配置等，然后 agent 更新 随 Docker 启动的默认配置，启动 Kylin 服务、监控等。在这个过程，可以通过配置化方式实现 Kylin 节点、集群的动态伸缩。基于服务发现的集群管理在测试中，不久将提交社区。

此图是简化后样例，包含：

1. 集群名称。
2. 集群的 servers 列表列出集群的 IP 集合。
3. 覆盖默认的集群配置，实现集群间的差异。比如滴滴有两个 HBase 集群：0.98 和 1.4.8，Kylin 集群使用哪个 HBase 集群可以配置在覆盖属性这里。
4. build & query & status 节点的组成 IP 集。

3. 负载均衡

集群的负载主要是：查询的压力，构建压力。然而查询和构建是和 Cube 绑定的，所以只要能实现 Cube 在集群间的流转就可以控制查询与构建的压力。只要实现了 Cube 的集群间流转就实现了查询与构建压力的流转，然后结合集群的负载监控以及 Cube 的流转调度，就可以实现集群间负载的相对平衡。

实现思路：我们实现 Cube 集群间流转，核心点就是当创建 Cube 的时候如何决定 Cube 创建到哪一个集群上，之后查询和构建的压力就落在该集群上。

具体实现：

1. 当有 Cube 创建需求的时候，就会把这个请求存储到一个有序队列。
2. 然后开放平台会消费队列中的 Cube 创建请求转发到 Kylin 集群去创建 Cube。
3. 配置中心记录 2 条核心配置：1）用户可以在哪些集群创建 Cube；2）各个集群接收 Cube 的占比是多少；
4. 开放平台读取配置中心的配置，经过简单的计算之后就能确认该 Cube 创建请求应该发送到哪个集群。

举例：基于用户与集群的配置关系可以实现用户的资源隔离。基于集群接收 Cube 的占比，比如 A 集群 ：B 集群 = 1 ：3，那么如果有 4 个请求的话，开放平台就会向 A 创建 1 个 cube，向 B 创建 3 个 cube。当过一段时间之后，可以根据 A、B 集群的负载在配置中心修改 A 集群与B集群的比例，实现流量的转移。

为什么要设计 Active/Standby status 节点？

1. 因为 Kylin 所有的元数据都是在节点启动的时候加载到内存，在此之后对任何元数据的变更，都是通过广播的形式告诉其他节点发生了什么事情。接收到广播的节点，根据实际的需求来更新内存中的元数据，这样在广播失败或者更新效率慢的时候就会造成集群间元数据不一致的问题。
2. 然后 cube-model-table 的创建会有依赖关系的，比如创建 Cube 的时候，当前节点内存中必须要 Model 的信息，如果广播没有完成立刻换一个节点执行 Cube 创建会报错。

基于以上两个原因就抽取单独的节点 status 作为 API  和集群交互的媒介。为了防止单点故障实现了 Active/Standby。除了 table&model&cube 的 CRUD 外，集群内还有其他的元数据广播。实验发现 1500+ Cube 的集群使用元数据操作串行的方式比随机广播的方式操作失败的比例大概减少 30%，所以我们在后来的实践中就改成一个按请求队列模式，实现集群状态节点管理。

字典改造

下面讲一个实际案例，滴滴对于字典的改造——全域字典。

1. 如何改造

首先，简单介绍下 Kylin 默认支持精确去重的字典是全局字典，限制是只能实现 Cube 内的复用。适用方式是设定计算指标为 “count_distinct”，选择 Return Type为”precisely (More Memory and Storage Need)”, 并在”Advanced Dictionaries”处选择“Builder Class ：org.apache.kylin.dict.GlobalDictionaryBuilder”。

通过名字可以了解到，滴滴这边实现的字典是可以支持跨 Cube 使用，所以叫”全域字典”。演变过程分两步:

第一步，MR-Hive 字典实现字典编码外置 Hive 表；

第二步，MR-Hive 全域字典改造实现 Cube 间依赖。

2. 案例

结合一个案例来说明下，如上图所示，是一个效果评测平台：

1. 圈定一部分标签数据；
2. 投放前分析数据在过去几个月的历史数据；
3. 投放后持续跟进效果。

在此过程中进行效果评估的数据是固定的，但是需要计算留存、用户数等指标，所以用到 Kylin 的精确去重指标。可是字典只需要构建一份就可以了，其他的数据可以公用这一份数据，这样就可以缩减 Kylin 的构建时间并减小 Kylin 的 Job 节点构建的压力。

为什么可以实现 Kylin 字典的外置？一起梳理下基本流程和关键核心点：

1. Kylin 的上卷、下钻的的精确去重得益于其实现的全局字典。
2. Build Dimension Dictionary 过程是将 uid 编码成 int 数值的过程，此过程是在 Job 节点单机完成的。
3. Build Base Cuboid 这步骤主要做两件事情：1）将原始数据的 uid 替换成编码后的 int 数值。2）做一个全维度（cube 设置的全部维度）的 group by 操作，此时 count_distinct 函数的计算结果会转化成一个 bitmap(RoaringBitmap)，此对象可以做或、与的按位操作。
4. 在 Kylin 返回给可客户端的时候只需要做一个 bitmap(RoaringBitmap).count() 操作，就可以得到结果。

我们从上边的流程可以分析出我们需要做的事情有两个：

1. 模拟Build Dimension Dictionary过程，充分利用计算集群的能力实现 uid 与 int 编码的过程，并将结果存入 Hive。
2. 改造 Build Base Cuboid 过程，实现高效的字典替换。

3. 效果

上图简单总结了做全域字典的初衷、改造的可行性思考、取得的效果。

用上图展示下实际的使用效果，右边是默认的全局字典的配置后的元数据展示，左边是全域字典的配置形式。可以看到字典在 reuse 的时候其实引用的是其他 cube 的列，同时也设定了引用的 model 和 cube，从截图可以看出它们来自不同的 cube。

Kylin 本身是不提供调度依赖的，但是实际企业生产是要有依赖关系的。滴滴的做法是在 cube 的字典构建完成时，在 hdfs 记录一个 success 信息，依赖此字典的 cube 把此依赖配置到当前 cube 的依赖，这样就从企业生产环境的角度实现了 cube 间调度的相互依赖。

此处列举几个做这件事情的外因：

1. 任务构建对集群的压力：如果不使用全域字典，则每个 cube 都需要构建一遍字典，这样任务的构建时间就会变长。job 节点的并发有限会造成任务的等待时间变长，字典构建这一步是在 job 节点单机完成的，对 cpu 和网络的压力也比较大。
2. 影响 Hadoop 集群的稳定性：全局字典是以文件的方式存在于 HDFS 上的，使用的是AppendTrieDictionary，实际进行判定字典插入和查询是按照 Slice 为单位进行交互判定的。所以当字典比较大并且事实表的纪录条数比较多时，就会频繁的从 HDFS 获取数据对比，这时候网络压力就会很大。上图为滴滴一个构建 cube 启动后 hadoop 集群网络的飙升情况。

MR-HIVE 字典的实现思路：

1. 创建一个 kylin-flat-table 的 group_by 表，将配置的全域字典列做一个 group by 操作，写入对应的分区中，因为是字典所以保留一条纪录即可；
2. 创建一个 cube 的 global_dict 表，设计的数据结构为 key、dict value、partition；
3. 将 global_dict 表和 group_by 表做一个增量计算则可以获取到增量字典，完成 global_dict 的编码工作；
4. 将 kylin-flat-table 表和 global_dict 关联，替换 kylin-flat-table 中的 uid 为编码后的 int 字典值；
5. 根据之前讲的 Build Base Cuboid原理，就是实现了全域字典外置 Hive 字典的实现。

此图是 global_dict 表和 group_by 表做一个增量计算的 SQL。当字典值比较大的时候，row_number() 排序的效率很低，构建会长时间卡住。经过分析发现是全局排序的问题，那有没有解决办法呢？

经过调研发现，Hive 在实现全局 order by 的时候并没有这样的性能瓶颈。调研发现其使用的是 Hadoop 提供的 TotalOrderPartitioner，因此借鉴此实现优化了 row_number() 的性能瓶颈

基于全域字典的外置 Hive 特点，在滴滴延伸出以下几个场景：

1. Hive 字典已经外置，那么字典的构建可以通过 ETL 来实现，只需要满足global_dict字典表的结构即可，不需要再 kylin 流程中构建了。
2. 在以往的用户分析中，用户只能分析出用户的结果数。经过此改造后结合滴滴实现的count_distinct_detail函数，分析人员不仅可以知道用户数是多少，并且知道是谁，从而可以和广告营销等系统打通。
3. 字典外置可以容易的实现 cube 间的依赖，让数据的组织更加合理、高效。