Sophon ：Hulu智能OLAP缓存层技术实践

520jefferson 2020-12-08

展开全文

内容来源：DataFunTalk

导读：本次演讲内容主要分为四部分：第一部分介绍Hulu的大数据架构；第二部分介绍Sophon，Sophon是一个轻量的OLAP中间缓存管理层，它的下层主要依赖于OLAP计算引擎—Impala；第三部分介绍 Impala在Hulu的一些应用实践；最后是一个全文的总结。

Hulu的大数据架构

Hulu大数据架构分成4部分，架构最底层有两个自建的数据中心，它主要用于公司内部的大数据平台，包括大数据的存储，离线和实时计算，以及OLAP查询引擎等。同时也有一部分的data和data mart存储在AWS的S3上，使用一些Cloud Native的查询引擎，例如snowflake等；基础设施层的上层是一些数据团队，包括广告，Metrics团队等，最上层提供一些广告售卖和一些报表服务等。

Hulu大数据的基础技术架构如上图，整体上是比较拥抱开源的，数据采集使用的是Flume、Kafka，数据存储用的是HDFS和HBase，资源调度用的是yarn，离线计算用的是Spark、MapReduce、Hive等，在线计算主要用Impala、Presto以及一些自建的引擎，Presto是一个FaceBook 开源的MPP架构OLAP查询引擎，在Hulu主要用于Ad Hoc的查询，Impala是Cloudera开源的MPP架构OLAP查询引擎，主要服务于一些自建的数据仓库，我们自研的主要是Nesto和Sophon，Nesto是一个能够查询嵌套数据结构，基于MPP架构的查询引擎，Sophon是一个OLAP缓存的管理引擎，同时提供查询和路由的功能，再往上就是统一的客户端管理，包含统一的配置管理，自动更新等。我们的监控，除了采用Cloudera的自带监控外，还有自研的监控组件Hawkeye，主要是对Cloudera监控的补充，它可以用来统计整个HDFS和各种计算引擎的指标。

Sophon

1.广告场景

在介绍了整个Hulu的大数据平台架构以后，我们来具体看一下今天要演讲的Sophon。首先介绍下Sophon的研发背景，Hulu广告的数据集市比较复杂。Hulu主要以一些质量比较高的电视剧为主，所以它的广告是受到很大限制的，主要以展示类广告为主，这种广告一般是提前售卖的方式，它不是一种实时竞价的方式，提前售卖意味着今年10月份可能去谈明年一年广告售卖的单，这个单会非常非常大，在明年一年的投放中会细分成小的订单，它会整合到广告售卖订单系统中，售卖周期会很长，预先售卖的方式不可能把广告售卖的十分完整，总会有一部分广告位或多或少，是通过实时竞价的方式售卖出去的。因此整个系统会产生很多的事实表，包括广告的投放，广告的交易等。这些事件都会形成很多fact事实表，以及非常多的dimension，包括订单的具体信息，用户的具体信息，年龄，所在的州等，这样就会形成很多的事实表和维度表在你的数据集市Data Market中，如果用一个传统的方式对上层提供服务的话，那么你是需要一个很长的pipeline来处理这些数据。

同时对于不同的事实表和不同的领域，需要建立不同的数据集市，有时比如你关心广告的售卖情况和具体的投放的关系时，就需要用相关的事实表和维度表来形成特定的数据集市，不同的数据集市之间可能复用了部分相同的表，导致它们之间存在一定程度的耦合关系，这样当维度表的关系一旦发生修改的话，就意味着你上层的在不同pipeline的逻辑都需要修改，整个处理流程非常长。还有就是需要由不同领域技术背景的人去服务你的各个层次，因此我们希望这个过程可以更加一致，更加统一，更加灵活。再有就是来自bussiness的需求，比如当数据分析师用到一些新的维度组合时，你原有的数据集市可能满足不了，那么当你去重新建立一些pipeline去满足这个新的需求时，时间周期可能会非常久。所以我们考虑能不能把这些东西统一起来，把所有的数据关系集中到一起，当你有新的需求出现时，我能利用现有的模型和数据，在一个适当的响应时间内返回结果，然后系统能自动的优化cache，加速查询。在这个方面我们尝试做了改进，把由下到上的过程统一到一起，把它结合到一个过程，就省掉中间各种流程。因为本质上那些数据之间的关系不随数据集市的不同而改变，既然不变的话，就能够构建一套系统在知道这套关系以后能自动的根据上层的需要去生成对应的数据集市需要的数据，直接去服务于用户，因此这就是我们尝试做Sophon的初衷。

2.Sophon的框架

从Sophon整个的框架可以看到，它上层对接的是Bussiness UI工具，比如MSTR和Tableau，而对于数据建模人员来说，它提供一个界面去编辑和配置数据之间的关系是什么。它只去管理和重写SQL，不会真正去做任何的查询,这样可以保证它是一个非常轻量的引擎。真正的查询工作它会下推给Impala或者其它一些数据引擎去实现。那么有一个问题，当我的数据表很原始的时候，我把很大的fact表接过来，查询速度会很慢，需要对整个过程进行加速，因此我们做了许多预聚合运算，存储了许多预聚合表，Sophon并不会把这些预聚合的表缓存到内部，而是直接存储为Hive表，且记录所有表之间的join关系。这一点和Kylin不同，Kylin将预聚合记过存储在HBase中，本身并不记录预聚合表与其他维度表之间是否能够join。虽然Kylin能够配置维度之间的函数关系，但是只能适用于有限的几种情况。另外，Kylin只支持Snowflake这种单fact的模型，在这种情况下，它可以把速度提升到很快毫秒级别，但牺牲了它的灵活性。因此，Kylin和Druid更倾向于把我的查询提升到非常快的程度，达到毫秒级别，一旦没有命中的话，就会发送给后端的Hive引擎，这个速度就会有很明显的下降。对于Sophon来说，如果没有命中聚合表，但是select的维度和聚合表存在join关系，Sophon会把两者join在一起做联合查询。一般来说，聚合表的规模远远小于Fact表，当这个join下推到一个MPP架构的查询，它的优化还是很可观的，即使我们没有在完全命中缓存的情况下，一个十几分钟的查询在这种情况下会缩减到几秒到十几毫秒，也可以非常明显的被上层感知到，所以它的适用范围会比这些查询引擎广一点，但是它的缺点就是会比Kylin这些引擎慢一点，到不了毫秒级的查询，因此使用场景会有一些区别。

3.Sophon的建模

上面介绍了Sophon是做什么的，下面我们来介绍一下Sophon建模的一些细节。Sophon最主要的功能就是表与表之间的Join，下面针对Join给出一个例子，如上图所示，我们有一个Fact表和几个Lookup表join到一起，在lookup表中，我们配置了维度与维度之间的层次结构关系，类似传统数据库的函数依赖，比如Dim1是一个hour，Dim2是一个Month，你的FK可能是一个year，同时系统中预构建了一个聚合表，它只命中了维度Dim1和度量M1，然而，查询语句并没有完全命中聚合表，那么在一个Kylin或者其它的系统中就直接放到后端的查询引擎计算了，但是对于Sophon来说，可以把Lookup1和Lookup2结合起来，把这两个表join到一起，这样聚合表就可以在这种情况下使用，这样的另一个好处是，当Lookup2的Schema结构发生了变化，预聚合Aggregation1同样是完完全全可以用的，不需要去做任何的重建工作。当然Kylin也支持Derived Column的特性，但是有很多的限制，像这样的例子无法适用，由于Sophon所记录的原始表之间的关系中，join链往往比较长，很多情况都无法通过Derived Column这种方式来优化。

另外一个功能是，我们希望聚合可以自动构建。聚合本身就是一个计算Cube的过程，在Kylin里面，决定聚合哪些维度组合，需要人工探索，尤其是当Dimension维度规模比较大的时候，Cube膨胀的规模会变得非常大，但是Sophon所遇到的场景本身维度就非常多，当你想通过人工去找出里面的规律，会非常困难。在Sophon里面，我们通过使用一个贪心算法去决定去建哪些聚合，具体来说就是每当我们选择一个节点物化一个维度组合生成缓存的时候，它对查询Sophon的所有query都会产生一定程度的优化（没有被优化的认为优化程度为0），总体的查询代价就会减少，因此只要我们建立一个模型来计算查询的代价降低了多少，并且每一步选择一个代价减少最多的节点计算，就可以逐步通过这种贪心的方式得到一个比较优的结果。这种方法已经被证明与最优解之间的差距在一个有限范围内。综上可以看出，如何选择代价是一个关键的问题，直观上讲，代价应该和join的规模成正相关，当你join的两张表越大，代价就越大; 如果聚合表的规模很大，那么就会耗费更多的存储空间，查询优化的效果也不经济，这种情况需要引入额外的惩罚。最后的问题是，这个算法本身的复杂度跟cube的复杂度是成线性关系的，假如说有n个维度，那么复杂度将是O(2^n)。为了减少这个规模，我们一般需要去做一个聚类，来确定线上哪些维度是经常在一起查询的，然后再根据这样一个贪心的算法，去进一步的找出我们物化哪些节点，使得最终的代价最小，收益会最大。

4.Sophon与后端引擎的联系

最后是如何将Sophon和后端引擎融合到一起，这方面最重要的是将如何根据预先定义的模型选择聚合表和join路径与如何对输入的查询进行改写和优化分离开来。可以从下图看到，首先系统将输入的原始SQL翻译成逻辑执行计划，然后从逻辑执行计划中提取具体是要查询哪个Cube，哪些dimension和Measure信息。然后将这些信息传递到右侧的优化系统中，这一部分与SQL采用什么样的语法没有任何关系。之后，系统根据这个信息去判断应该怎么去展开去命中需要的缓存，是否能命中聚合表，根据不同的选择去决定怎么把这个表展开，展开之后再去决定在这种场景下有哪些优化规则可以用的上，然后通过一个适配器生成改写SQL的规则，之所以这样做是希望Sophon可以不仅支持单一查询引擎，尤其是在当前背景下，希望以后可以支持一些Cloud native的引擎，比如S3上的Snowflake引擎，目前的趋势就是，以后的一个查询，一部分1年以内的数据可能在In-house的集群里，一年以前的数据在云上，我希望可以自动将这两部分数据的查询结果结合起来，这个时候可以由Sophon对这两部分数据进行管理和Merge结合起来，希望上层看到的是一个统一的视图，具体的底层细节数据是在云上还是在云下对于上层用户是无感知的。

5.Sophon在广告数据集市的数据规模

下面介绍下Sophon在广告数据集市的数据规模，有5张事实表，50+维度表，超过500个维度，数据量压缩以后有50TB的数据，大概是10倍的压缩比例，压缩格式是ORC，这个是一个数据集市的量，目前我们要解决的场景就是这个，大部分的查询在秒级或者十几秒内。

简单总结下Sophon的特点：首先，它很轻量，结构简单。其次，它比较智能，它会自动的帮你完成很多事情，数据工程师只需要去关注建模，之后它便能够根据用户的查询需求来选择应该计算哪些中间结构去加速查询，并自动缓存数据。再次，它比较灵活，它不仅可以直接命中缓存表，还可以将缓存表与其它的表进行关联，由于缓存表的规模一般比较小，这样的场景下，它的速度也可以有明显的提升。最后，它易于集成，由于分离了优化模块和SQL改写模块，未来可以集成更多查询引擎。

Impala在Hulu的应用

1.Impala的选型背景

除了Sophon以外，Sophon的底层非常依赖Impala,下面我们介绍下Impala在Hulu的应用，首先介绍下Impala的选型背景，Impala是一种MPP查询引擎，它的查询速度相对Hive、Spark要快很多。其次它对Hive SQL的兼容性非常好，数据集市的迁移会非常方便。再有它是C++编写的，OLAP在很多情况下Worker节点之间的Shuffle数据量会非常大，对于java-based的OLAP查询引擎的话，GC垃圾回收的压力非常大，很容易导致Full GC，所以Impala在内存管理这方面会友好一些。再有就是它向上提供的JDBC或者ODBC的操作，由于这些特性，我们选择了Impala作为我们的后端。

2.Impala的集群

Impala在Hulu有70多台机器节点，主要有两个集群，广告的是一个单独的集群，Hadoop的机器节点有1千多台，存储了Hive中的1万多张表，支持的文件格式有parquet、orc、text、json、sequence file。Impala默认是不支持ORC格式的数据查询的，这也是很多时候用Impala查询hive 表的痛点，而Presto支持的数据格式就会丰富一点，因此我们在这方面做了单独的支持，再有Impala有一个问题就是它的元数据是会被Cache缓存的，被Cache的话，它不能及时的知道Hive层的数据变动，而我们的数据导入 pipeline入库的过程一般来说还是通过Hive或者Spark这些工具导入进去的，那么怎么在数据导入以后，让数据有及时的感知，这是个很重要的问题。

由于ORC在hive表中应用广泛，Hive中90%的表都是ORC格式的。因此如果要使用Impala，我们有很强的需求使它能够查询ORC表，Hulu在17年开发了这个Feature，在C++这一层实现了Impala的scanner，目前它能够支持查询主要的数据类型，包括一些比较简单的数据格式的下推，这些工作由另外一个同事完成，目前已经反馈给社区，这个Feature会在2.13和3.1的版本中会被Release出来，现在正在开发复杂类型的支持和条件下推，还在和社区讨论过程中。

3.元数据的更新

另外一个问题是关于元数据的，当你有新的数据进来，或者表的Schema发生了改变时，Impala无法自动更新这些变化，查询在执行过程中出错。这种情况下，简单的办法是根据Impala提供的指令手动更新元数据。但是我们希望这个过程能够更加的自动化，来减少人工的参与。

对于这个问题，我们在Impala外部构建一个Pipeline直接拿到HDFS和Hive的audit日志形成一条事件流，通过Kafka灌入Spark Streaming中，来监控数据和表schema的变化，当变化发生时，会自动触发Impala的刷新，这是一个独立的外部组件，同时也有一个报表系统，会具体记录用户对集群做了哪些操作，如果这个时候有人误删了一些表或者一些有害性的操作，报表系统会检测出来，并触发报警。