Doris基于Hive表的全局字典设计与实现

首页版块访问AI主站注册发帖

三寸九州发布于2021-03 浏览:3765 回复:2

快速回复

感谢7月25日来参加Doris与Dolphin Scheduler合作举办的线上Meetup的小伙伴们，现在为大家带来Meetup的内容。本次Meetup请到了来自百度、奇安信、美团点评和易观的技术大牛带来技术分享。了解更多详情请关注Doris官方公众号。嘉宾分享回顾会陆续放出，公众号后台回复“0725”立即get回放录像及嘉宾PPT。

本次为大家带来的是第二期内容回顾：《Doris基于Hive表的全局字典设计与实现》，第一期在这儿。

本期主讲人：王博
美团点评

数据开发工程师

讲座主要分为四部分内容：

为什么需要全局字典

什么是全局字典

全局字典的技术实现

总结与未来规划

为什么需要全局字典

我们在业务上需要OLAP引擎能够⽀持精确去重的预计算，但这个需求本身是不是合理的呢？

目前线上数据分析会有一些与⾦钱相关的计算，比如某一个业务方要计算他们个人的销售业绩、以及一些广告的点击率等，这些能够转化为金钱的计算需求是要求精确去重的，近似去重是⽐较难以容忍的。

目前Doris的BITMAP列已经⽀持了对于整型的精确去重计算，但是现实的场景是业务会有更多的数据类型需要精确去重，比如String，这些非整型的数据类型是无法直接存入BITMAP的。

什么是全局字典

1. 全局字典的数据结构

那么基于以上的问题我们需要引入一个新的数据结构——全局字典，其实就是一个比较大的MAP。

MAP的KEY列保存了原始值，VALUE列保存了编码后的整型值。这样的话，即使我们有一个字段是String类型的，如果能将String字段值转成一个整型值，再把原始值的String类型的值和转换后的整型值都存入这个MAP，那么在导入数据时就可以读取这个MAP，先将原始值替换为整型值，再将整型值导入到Doris的BITMAP列中，最终实现对String类型的精确去重查询。

其实就是设计一个中间的映射数据结构去实现对原始值的转化，再把转化后的值存入Doris，这样就能实现对任意数据类型进行精确去重。

2. 全局字典的构建与使用流程

当然，只有一个MAP的数据结构是不够的，它需要一个完整的使用流程。

简单来说，首先是更新，这个更新就是对全局字典数据结构进行更新。第二步是读取，在编码的阶段，也就是已经有了字典之后，需要读这个字典的数据，然后根据映射把原始值转成编码值，核心的步骤就是一个构建和编码的流程。这两个流程是比较容易形成性能瓶颈点的，这里后面会具体说到。

3. 业界目前主要有两种全局字典的实现

第一个是基于Trie树的全局字典，第二种是基于Hive表的全局字典。今天我们主要讲的就是基于Hive表的全局字典。

全局字典的技术实现

1. 基于Trie树的全局字典

这个方案主要是Kylin引擎支持，在美团内部有大规模的落地经验，大部分业务都用的这套方案。这套方案在内部使用的优点缺点也是很明确的了。

（1）Trie树的基本设计

Trie树就是基于前缀匹配的树。每次插入的时候都会进行一次前缀的匹配，如果前缀能匹配上，多出来的部分会做一个大小的比较。如果新插入的值大于已经存在的值，就插入到右侧，否则插入到左侧，会做一个排序和比较的过程。如果前缀都不匹配的话，会生成一个新的节点。

根据Trie树的特性我们可以得出一个结论——Trie结构在数据写入的时候就已经完成了去重和排序的工作（其实我们主要利用它去重的特点）。在读取数据时，数据在Trie树中的位置就是对应的编码值。也就是说，当数据写入完成时就达到了我们的预期——去重以及对数据的编码。

（2）对Trie树进行分片

在实际的线上环境中有一个Trie树是不够的，因为通常Trie树非常大，小的可能几十MB上百MB，大的几十GB上百GB都有，所以一定要做树的分裂操作。

一个树经过分裂会分成多个分片，这样做的好处是不用一次把数据全加载到内存中，可以通过对分片文件进行缓存从而提高访问效率。但是目前的方案容易在缓存中出现一个性能的瓶颈。具体实现时，一个子树就是一个分片，每一个分片都是位于HDFS的一个文件。

（3）Trie树总结与分析

优点是资源比较节约，写入一个树的过程目前是一个单节点的构建，而编码阶段只需要做一次MAPREDUCE操作就可以完成字典的读取和对原始值的编码，相对来说资源比较节约而且在大部分场景下性能表现良好。

但是我们经过内部使用，也发现了一些局限性。

主要问题是在缓存上，上面提到了要对Trie树进行分片，但是在两种情况下分片缓存的命中率会下降：

字典非常大。我们有些业务去重字段基数很高，所以字典很大，达到几十GB上百GB的情况，分片会非常非常多。虽然字典是排过序的，但是输入的数据可能是比较离散的。比如你有10个分片，输入10行数据，你要对这10行数据做一个编码，最差的情况是你要访问10次HDFS，比较好的情况是10行数据可以在一个分片中查到，生产环境中不能对用户的输入值做出最优的假设，事实上当字典基数很高时，最差的情况往往容易发生。

有多列高基数去重列。比如当你对一张表进行编码的时候，因为单节点的内存是有限的，每列都需要加载不同的字典，会导致字典不停地读入读出，因为内存不足最终也会导致缓存的命中率下降。这种下降带来的影响是字典整体的构建流程耗时变长。

第二个局限性是目前字典是单个Reducer构建的，写入HDFS时一半数据位于本节点，一半数据分散在其他节点。而在我们使用过程中观察到HDFS不会把网络带宽作为一个查询路由，你的查询有可能打在单个节点上从而把HDFS的单节点网络带宽打满。此外，单节点的构建效率也比较低。

2. 基于Hive表的全局字典

基于SparkSQL+Hive表构建字典这套方案的优点是可以解决Trie树的瓶颈场景。

这套方案其实非常简单，它只需要三张表，第一张表是字典表也就是MAP，KEY列保存了去重列的原始值，VALUE列保存了去重字段编码后的值，该列为整型。第二张表是用户输入表，就是用户那张原始的Hive表。还有一张是去重表，保存了每次输入的值去重后的集合。

（1）构建流程：抽取去重值

如上图所示，右上是一个原始表，保存了两列数据，我们期望对这张表进行的操作是，把name列替换成一个编码后的列。第一步是要把去重值抽出来，用insert和group by写一个SQL，将需要去重的值保存到一张grpby_table。

（2）构建流程：构建全局字典

第二步要做一个全局字典的构建，这步其实就是写入字典的过程。

看右侧，我们期望的结果是对上面这个需要去重的值的grpby_table的name列生成一个对应的VALUE，也就是进行编码。

步骤#1是获取本次导入的新增的去重值，用去重值表left join字典表，通过where条件可以把新增的值抽出来，因为我们只需要对每次新增的去重值做一个编码。

步骤#2是使用row_number函数对新增的去重值做一个编码，现在我们已经拿到了这样一个集合。

步骤#3是和字典表做一个union，将增量值和历史值合在一起，结果就是右侧下方的表。

（3）构建流程：对原始值进行编码

有了字典之后我们就可以进行编码操作了，这个编码是对原始值做编码。

右上方左边的表是原始表（Hive表），右边是我们对name列编码后的字典，这样我们只需要用原始表的name列和字典表的dict_key（name）列做一个join就能得到右下所示的表。这就完成了对原始表进行编码的操作。

这里要关注一个问题，当去重列数据倾斜时容易出现性能瓶颈（后文会具体说明解决方案）。

（4）性能瓶颈分析：字典构建阶段

因为我们目前使用SparkSQL的窗口函数（row_number）对去重值做解码，此函数的瓶颈在于它是单机执行的，而非分布式的。可用内存有限，单次输入亿级别的行数时容易内存溢出。

这个问题是比较容易解决的：每次拿到去重值之后先进行一个统计计算，如果行数超过一亿行就做一个切分。

（5）性能瓶颈分析：编码阶段

第二个瓶颈相对来说不是很容易解决，当去重列发生数据倾斜的时候，要么跑不完要么耗时非常长。

第一种优化方法是使用MAP JOIN，但MAP JOIN也是有性能上限的。

第二种优化方法是，如果倾斜列空值较多那么就只对非空值进行编码，先把原始值进行切分，把非空的抽出来之后就一定不会倾斜了。

还有一种非常极端的情况，如果数值特别大无法使用MAP JOIN而且倾斜的值不是空值的，那么就对原始表，维表和字典表都加一个随机的前缀（这个问题其实就转化为如何解决Spark数据倾斜的问题）。这种方法的弊端是，添加了随机前缀意味着原始值和字典值都进行了膨胀。

总结与未来规划

1. 两种方案对比分析

对比两种方案，有两个基本的维度：字典构建和对原始值进行编码。

在字典构建上，基于Hive的方案因为是使用SparkSQL和窗口函数实现的，这种方法并行度很高，性能会很好；而Trie树是基于单节点构建的，相对来说吞吐量会低一点，会慢一点。

在编码上，基于Hive的方法一次Shuffle完成一列的编码，如果有多列Bitmap列的话，就需要多次Shuffle这时资源开销会大一些；而Trie树是不需要Shuffle的。

基于Hive的方法相对有更大的弹性，除了极端的数据倾斜的情况，Hive这个方案是几乎没有什么瓶颈，是很占优势的；但是Trie树的方案它的瓶颈不在资源上，这也是我们线上落地的时候比较重要的痛点，它会在特定情况在缓存上出现瓶颈，这点是加资源也无法解决的问题。

2. 美团为什么选择了基于Hive表的方案？

最终我们设计Doris的字典方案时选择了Hive表方案是因为考虑到：

（1） Hive表方案除了个别数据倾斜情况下整个方案没有什么明显的瓶颈，未来可以考虑从Spark引擎层的角度解决极端的数据倾斜问题。（2）Hive表方案可以解决Trie树的短板问题。（3）这个方案实现起来比较简单可以很快落地。（4）弹性比较好，在资源足够的情况下可以达到非常高的吞吐，很高的并行度。

3. 未来规划

目前在美团内部，基于Hive表的方案已经成为 Hive2Doirs 流程的默认方案，已经跑了大概2个月了。未来计划需要有更多的业务接入，可能会面临更复杂的考验，我们也期待对这套方案做持续的改进。

未来优化的方向是：

（1）考虑从Spark的层面解决数据倾斜的问题，比如引入SparkAQE特性（当Join时发现单个节点处理数据量过大时，会把数据做一个分发，引入多个节点去处理倾斜的数据，这样就可能会解决数据倾斜的问题），未来如果引入这一特性，整套 SparkSQL+Hive 做字典的方案将会没有明显的瓶颈。

（2）使用Trie树与Hive表融合的方案，这个方向是一个很理想的设计，但是改造成本会非常高。在大部分情况下使用Trie树做字典的编码，在基数高字典大的情况下就把Trie树转成Hive表然后使用Join完成编码。不过这套方案的改造成本比较高，比如写入时Trie树支持分布式构建，读取时需要能转成SparkRDD再转成Hive表。

欢迎扫码关注：

Apache Doris(incubating)官方公众号