17 从Dremel到Parquet（二）：他山之石的MPP数据库

你好，我是徐文浩。

在上节课里，我们看到了Dremel这个系统的数据存储是怎么回事儿的。不过，只是一个支持复杂嵌套结构的列存储，还没有发挥Dremel百分之百的威力。像Hive也在2011年推出了自己的列存储方案RCFile，并在后续不断改进提出了ORC File的格式。

列存储可以让一般的MapReduce任务少扫描很多数据，让很多MapReduce任务执行的时间从十几分钟乃至几个小时，下降到了几分钟。更短的反馈时间，使得数据分析师去探索数据，根据拿到的数据反馈不断从不同的角度去尝试分析的效率大大提高了。

不过，人们总是容易得陇望蜀的。当原先需要花几天时间写MapReduce程序才能分析数据的时候，我们希望能够通过写SQL跑分析数据。当原先SQL运行要30分钟、一个小时的时候，我们通过列存储把SQL执行的时间缩短到5分钟。但是在这5分钟里，我们的数据分析师该干嘛呢？只能去倒杯咖啡发个呆么？所以，我们自然希望SQL在大数据集上，也能在几十秒，甚至是十几秒内得到结果。

所以Google并没有在列存储上止步，而是借鉴了多种不同的数据系统，搭建起了整个Dremel系统，真的把在百亿行的数据表上，常见OLAP分析所需要的时间，缩短到了10秒这个数量级上。那么，这节课我们就来看看Dremel是通过什么样的系统架构，做到这一点的。

和所有工程上的进展一样，Dremel也是从很多过去的系统中汲取了养分：

第一，它从传统的MPP数据库，学到了数据分区和行列混合存储，并且把计算节点和存储节点放在同一台服务器上。
第二，它从搜索引擎的分布式索引，学会了如何通过一个树形架构，进行快速检索然后层层归并返回最终结果。
第三，它从MapReduce中借鉴了推测执行（Speculative Execution），来解决了少部分节点大大拖慢了整个系统的整体运行时间的问题。

而这三个的组合，就使得Dremel最终将百亿行数据表的分析工作缩短到了1分钟以内。

通过这节课的学习，我希望你不仅能够学到Dremel的具体架构的设计，更能够学会在未来的架构设计工作中，博采众长，做出让人拍案叫好的系统设计。

瓶颈并不出现在硬件

Dremel采用的列存储，已经极大地减少了我们扫描数据的浪费。在论文里，Google给出了这样一组数据：在3000个节点上，查询一个87TB、一共有240亿条数据的数据集，查询的内容是一个简单的Word Count程序。如果采用MapReduce去读取行存储的数据，那么需要读取87TB的数据。而如果采用列存储的话，因为只需要读取一列数据，所以只扫描了0.5TB的数据，整个MapReduce程序执行的时间，也缩短了整整一个数量级。

这也是为什么，在Dremel论文发表之后，开源社区很快跟进了这个支持嵌套的列存储的存储格式，也就催生了Parquet这个开源项目。

不过，如果你去看论文中的图10，你会发现，使用Dremel比传统的MapReduce读取列存储的数据还要再快一个数量级。那这又是怎么做到的呢？请你和我接着一起往下看。

我们之前提过很多次，MapReduce虽然伸缩性非常好，非常适合进行大规模的数据批处理，但是它也有一些明显的缺陷，其中很重要的一个问题，就是每个任务都有相对比较大的额外开销（overhead）。

所以即使有了Hive，可以让分析师不用写程序，可以直接写SQL，另外列存储也让我们需要扫描的数据大大减少了，但是MapReduce这个额外开销，始终还是会让我们的分析程序的运行时间在分钟级别。

而前面的图里我们可以看到，Dremel则可以让我们这样的SQL跑在10秒级别。说实话，刚看到这个数据的时候，我是有点难以置信的。事实上，不只是我这样的工程师这么想，著名的Wired在Dremel发表之后就报道过，像Berkeley的教授阿曼多·福克斯（Armando Fox）就说，“如果你事先告诉我Dremel可以做什么，那么我不相信你可以把它做出来”。

If you had told me beforehand me what Dremel claims to do, I wouldn’t have believed you could build it.

不过，回过头来，从硬件的性能来说，这看起来又是完全做得到的。论文里给出的实验数据里，是用3000个节点，去分析0.5TB的数据，这意味着每个节点只需要分析167MB的数据。即使是传统的5400转的机械硬盘，顺序读写的确也只需要数秒钟，再加上网络传输和CPU的计算时间，的确也就是个10秒钟上下的时间。

Dremel系统架构

Dremel之所以这么快，是因为它的底层计算引擎并不是MapReduce。Dremel一方面继承了很多GFS/MapReduce的思路，另一方面也从传统的MPP（Massively Parallel Processing）数据库和搜索引擎的分布式检索模块，借鉴了设计思路。其实它的核心思路就是这四条：

第一点是让计算节点和存储节点放在同一台服务器上。MPP数据库和搜索引擎的分布式索引的架构也是这样的。
第二点是进程常驻，做好缓存，确保不需要大量的时间去做冷启动。这一点，也跟MPP数据库和分布式索引采用的架构和优化手段类似。
第三点是树状架构，多层聚合，这样可以让单个节点的响应时间和计算量都比较小，能够快速拿到返回结果。这个架构，和搜索引擎的分布式索引架构是完全相同的。
最后一点则仍然来自于GFS/MapReduce，一方面是即使不使用GFS，数据也会复制三份存放到不同的节点。然后在计算过程中，Dremel会监测各个叶子服务器的执行进度，对于“落后”的计算节点，会调度到其他计算节点，这个方式和MapReduce的思路是一样的。更进一步的，Dremel还会只扫描98%乃至99%的数据，就返回一个近似结果。对于Top K，求唯一数，Dremel也会采用一些近似算法来加快执行速度。这个方法，也是我们在MapReduce中经常用到的。

那么下面，我们就对着论文中Dremel的系统架构图，一起来看一下它是如何组合GFS/MapReduce、MPP数据库，以及搜索引擎的系统架构，来实现一个能够在数十秒内返回分析结果的OLAP系统的。

Dremel采用了一个多层服务树的架构，整个服务树里面有三种类型的节点：

首先是根服务器（root server），用来接收所有外部的查询请求，并且读取Dremel里各个表的METADATA，然后把对应的查询请求，路由到下一级的服务树（serving tree）中。
然后是一系列的中间服务器（intermediate servers），中间服务器可以有很多层。比如第一层有5个服务器，那么每个服务器可以往下再分发下一层的5个服务器，它是一个树状结构，这也是服务树的这个名字的由来。我们所有查询Dremel系统的原始SQL，每往下分发一层，就会重写（rewrite）一下，然后把结果在当前节点做聚合，再返回给上一层。
最下面是一层叶子服务器（leaf servers），叶子服务器是最终实际完成数据查询的节点，也算是我们实际存储数据的节点。

光这样讲系统的架构实在还是太抽象，我们还是来看看论文里给到的SQL的例子：

SELECT A, COUNT(B) FROM T GROUP BY A

这是一个我们在日常数据分析中很常见的SQL，它是从某一个表里T，按照某一个维度A（比如国家、时间），看某一个统计指标B（比如页面访问量、唯一用户数）这样的数据。这个SQL在Dremel上执行的过程是这样的。

首先，SQL会发送到根服务器，根服务器会把整个SQL重写成下面这样的形式。

SELECT A, SUM(c)FROM ( $R_{1}^{1}$ UNION ALL … $R_{n}^{1}$ ) GROUP BY A

其中的每一个 $R_{1}^{1}$ … $R_{n}^{1}$，都是服务树的下一层的一个SQL的计算结果，那么下一层的SQL是这样的：

$R_{i}^{1}$ = SELECT A, COUNT(B) AS c FROM $T_{i}^{1}$ GROUP BY A

这个解决办法其实一看就能看懂。因为原始的SQL是进行统计计数，那么我们只需要让中间服务器，分别去统计一部分分区数据的统计计数，再把它们累加到一起，就可以拿到最终想要的结果1。这里的$R_{i}^{{1}$就是对应中间服务器的中间结果，$T_{i}}$就是对应分配给当前中间服务器，需要计算的数据的分区。

事实上，这里面的$R_{i}^{1}$可以再用根服务器重写SQL的方式，进行再次重写，再往下拆分，我们可以有两层、三层乃至更多层的中间服务器。而到了最后一层，分发给叶子服务器的时候，就不能再往下分发了，叶子服务器会在它所分配到的分区上，执行对应的SQL并且返回。

行列混合存储的MPP架构

上节课我们学习过列存储的内容，我们知道Dremel的列存储本质上是行列混合存储的。所以每一个节点所存储的数据，是一个特定的分区（Partition），但是里面包含了这个分区所有行的数据。这样当数据到达叶子节点的时候，叶子节点需要执行的SQL只需要访问一台物理服务器。在这种情况下，我们可能有两种方案：

一种是对应的数据，就直接存放在叶子节点的服务器的本地硬盘上。这种方式，也是传统的MPP数据库采用的方式，也是Dremel系统在2006年，在Google内部开始使用的时候采用的方式，直到论文发表的2009年，这还是Dremel系统主要采用的方案。
另一种方式，则是叶子节点本身不负责存储，而是采用一个共享的存储层，比如GFS。Dremel从2009年开始，就逐步把存储层全部迁移到了GFS上。

把数据存储和计算放在同一个节点，以及将用户SQL查询重写，并行分发到多个节点并且汇总所有节点的查询结果，是MPP数据库的常见方案。这也是为什么Dremel论文里说，它从MPP数据库里借鉴了很多解决问题的思路。

树形分发的搜索引擎架构

而这个一层层服务树分发的机制，则是借鉴了搜索引擎的分布式检索机制。数据分区到不同的叶子节点上，就是相当于我们把不同的文档分片到不同的索引分片服务器上。

每一个索引分片服务器，会完成自己分片数据上的检索工作，然后把结果返回给上一层的中间服务器。中间服务器也会在自己这一层，把检索结果再进行合并处理，再往上一层层返回，直到根服务器。

我们可以拿一个例子来看看，Dremel和搜索引擎的分布式索引有哪些相像之处。最合适的一个例子，就是求一个数据集中排序的Top K，也就是前K项的返回结果，它对应的SQL就是这样的：

SELECT A, B, C FROM T ORDER BY D LIMIT K

然后这个查询，在根服务器就会被重写成这样：

SELECT A, B, C FROM ( $R_{1}^{1}$ UNION ALL … $R_{n}^{1}$ ) ORDER BY D LIMIT K

里面的每一个 $R_{1}^{1}$ … $R_{n}^{1}$，都是服务树的下一层的一个SQL的计算结果，那下一层的SQL是这样的：

$R_{i}^{1}$ = SELECT A, B, C AS c FROM $T_{i}^{1}$ ORER BY D LIMIT K

然后每一个 $R_{i}^{1}$ 可以再用根服务器重写SQL的方式，进行再次重写，再往下拆分。也就是叶子服务器还是会获取自己分片数据的TOP K，每一层都会去归并下一层的返回结果，并再计算一次TOP K。

这个和搜索引擎的分布式索引的架构是完全一样的，唯一的差别是，搜索引擎计算TOP K的方式更加复杂一些，需要利用倒排索引，以及根据搜索的关键词，计算文档的一个“分数”来进行排名而已。

这个架构中最核心的价值，在于可以通过中间服务器来进行“垂直”扩张。并且通过“垂直”扩张，可以在计算量基本不变的情况下，通过服务器的并行，来缩短整个SQL所花费的时间。也就是通过增加更多的服务器，让系统的吞吐量（Throughoutput）不变，延时（Latency）变小。这个“垂直”扩张，并不是所谓的对硬件升级进行Scale-Up，而是增加中间层服务器，增加归并聚合计算的并行度。

因为实际扫描数据，是在最终的叶子节点进行的，所以这一层花费的时间和性能是固定的。如果我们没有中间服务器，而是所有的叶子节点数据都直接归并到根服务器，那么性能瓶颈就会在根服务器上。

根服务器需要和3000个节点传输数据，并在根节点进行聚合。而这个聚合又在一个节点上，只能顺序进行，即使每一个叶子节点返回的数据，在根节点进行数据聚合只需要20毫秒，那么我们也需要1分钟才能完成3000个节点的数据聚合。

而如果我们在中间加入中间层的服务器，比如，我们有100个中间层的服务器，每个服务器下面聚合30个叶子服务器。那么中间层服务器就只需要600毫秒完成中间层的聚合，中间层的结果到根服务器也只需要2秒，我们可以在3秒内完成两层的聚合工作。

当然，在实际的SQL执行过程中，我们还有叶子节点扫描数据，以及数据在叶子节点和中间层，还有中间层和根服务器之间的网络传输开销，实际花费的时间会比这个多一些。但是中间层，帮助我们把数据归并的工作并行化了。我们归并工作需要的CPU时间越多，这个并行化就更容易缩短整个查询的响应时间。

我们的叶子节点越多，叶子节点返回的数据记录越多，增加中间层就越划算。论文里的实验部分针对不同的SQL和不同层数的中间服务器做了各种实验，你可以去仔细看一看。

这里，我们可以来对照着看看实验部分里，两个SQL中的Q2和Q3：

Q2：SELECT country, SUM(item.amount) FROM T2 GROUP BY country
Q3：SELECT domain, SUM(item.amount) FROM T2 WHERE domain CONTAINS ’.net’ GROUP BY domain

其中，Q2是按照国家进行数据聚合，因为国家的数量很少，所以每一个叶子节点返回的数据量也很小。但是即使这样，在没有中间节点的情况下，因为根服务器要和3000个叶子服务器一一通信、聚合数据，花费的时间也要20秒。而我们只要加上一个中间层，所花费的时间立刻缩短到了3秒，但是要注意，这个时候即使我们再增加中间层，时间也无法缩短了。

而里面的Q3，是按照域名进行数据聚合。我们知道互联网上的域名数量特别多，在这个SQL中，最终一共会有110万个域名。没有中间层的时候，执行时间需要超过一分钟。增加了100个节点的中间层之后，时间就缩短了一半以上，而当我们在中间层再加一层，把整个服务器的树形结构变成1:10:100:2900的时候，执行时间能够再缩短一半，到15秒之内。

其实，这个树形垂直扩展的架构，也是搜索引擎能从无穷无尽的网页中，快速在几百毫秒之内给到你结果的核心所在。

来自MapReduce的“容错”方案

除了MPP数据库和搜索引擎之外，Dremel也没有忘了向自家的前辈MapReduce借鉴经验。我们刚才看到，Dremel的整个服务器集群也不小，实验里就动用了3000台服务器。那么一旦遇到这种情况，我们一样要面临“容错”的问题。

而Dremel和MapReduce一样，会遇到网络问题、硬件故障。乃至于个别叶子节点因为硬盘可能将坏未坏，虽然仍然能够读取数据，但是就是特别慢，这些它会遇到的问题，其实MapReduce里都遇到过。

所以，Dremel自然也就大大方方地借鉴了MapReduce和GFS里，已经用过的几个办法。

首先是虽然数据存储到了本地硬盘，也会有3份副本。这样，当我们有个别节点出现故障的时候，就可以把计算请求调度到另外一套有副本数据的节点上。

其次，是借鉴了MapReduce的“推测执行”功能，Dremel也会监测叶子节点运行任务的进度。在3000个节点里，我们总会遇到一些节点跑起来特别慢，拖慢了整个系统返回一个查询结果的时间。往往99%的节点都算完了，大家等这几个节点要等上个三五分钟。这些节点无论是在MapReduce还是Dremel中都会存在，我们一般称它们为“掉队者”（Stragglers）。

而Dremel和MapReduce一样，一旦监测到掉队者的出现，它就会把任务再发给另外一个节点，避免因为单个节点的问题，导致整个任务的延时特别长。

另外，在MapReduce里，我们最终还是要等待所有的Map和Reduce函数执行完才给出结果。而在Dremel里，我们可以设置扫描了98%或者99%的数据就返回一个近似的结果。

一方面，从Dremel的实验数据来看，通常99%的到叶子节点处理的数据是低于5秒的，但是另外的少部分数据往往花费了非常长的时间，甚至会到几分钟。另一方面，Dremel是一个交互式的分析系统，更多是给分析师分析数据给出结论，而不是生成一个用来财务记账的报表，数据差上个1%~2%并不重要。

小结

好了，到这里，Dremel的架构我们就学习完了，那我们就一起来总结一下吧。

可以看到，Dremel对于大数据集下的OLAP系统的设计，并没有止步于我们上节课所说的列存储。

通过借鉴MPP数据库，把计算和存储节点放在一起，以及通过行列混合的方式，Dremel完成了数据的并行运算，而且缩减了需要扫描的数据。通过借鉴搜索引擎的分布式索引系统，Dremel搭建了一个树形多层的服务器架构，通过中间层服务器进行数据聚合，减少了整个系统计算和返回结果的延时。

而通过借鉴MapReduce的容错机制，Dremel会把太慢的任务调度到其他拥有数据副本的节点里去，并且更激进地抛弃那些“掉队者”节点的数据，在只扫描了98%~99%的数据的时候，就返回结果，尽可能让每个SQL都能快速看到结果。

其实，从硬件层面的参数来看，Dremel能够在几秒乃至几十秒内，扫描240亿条数据中的几列数据进行分析，的确是做得到的。Dremel本身也没有发明什么新算法、新架构，而是通过借鉴现有各类成熟的并行数据库、搜索引擎、MapReduce搭建起了一个漂亮的框架，把大部分人眼里的不可能变成了可能。相信这一点，对于所有想要做架构设计的同学来说，都会有所启发。

思考题

最后，按照惯例，还是给你留一道思考题。

Dremel从2009年开始把数据存储，从叶子节点的本地硬盘迁移到了GFS上。那么，为什么一开始Dremel没有把数据存储就放在GFS上呢？放在GFS上，和放在本地硬盘上，分别会有什么好处和坏处呢？以及对于这些好处和坏处，我们又有什么应对方案呢？

欢迎你在留言区分享你的思考和答案，和其他同学一起交流，共同进步。

精选留言（11）

在路上 👍（25） 💬（1）
徐老师好，我想Dremel一开始把数据放在硬盘上，是因为当时“计算和存储分离”还不是大数据领域的主流思想，MPP数据库把计算和存储放在一起的思路，在过去证明是有效的，Dremel借鉴过去的成功经验是理所当然的。在Dremel 2020年的论文的第3.1节提到“At the time, it seemed the best way to squeeze out maximal performance from an analytical system was by using dedicated hardware and direct-attached disks. As Dremel’s workload grew, it became increasingly difficult to manage it on a small fleet of dedicated servers”。也就是说，那个时候大家都认为把计算和存储放在一起才是最佳的方法，但是随着数据规模和查询负载的增加，服务器管理越来越困难。在今天看来把数据放在GFS上，一定比放在本地好，但是这中间其实经过了很多优化，一开始的时候选择把数据放在本地是更好的选择，因为相关的技术都是很成熟的，把数据放在GFS上需要解决很多未知的问题。把数据放在GFS上有很多好处，第一，数据扩容方便，管理简单；第二，数据拥有多个副本，对容灾友好；第三，数据可以被Dremel之外的工具使用，也方便和其他团队共享。
2021-11-02

乐天 👍（10） 💬（0）
分开存储的好处：计算和存储分离，可以提供资源的利用率，数据量大就单纯增加存储节点，计算量大就增加计算节点，能更好的利用资源。同时任务调度时不用综合考虑节点的性能和数据的位置。坏处：增加了网络传输的时间。这样做是因为硬件性能特别是网络传输设备的提升很大，大数据量的传输已经不是大问题了，数据传输的时间可能比任务等待调度执行的时间还要短。
2021-11-23

陈迪 👍（3） 💬（0）
尝试回答思考题：采用GFS最明显的好处是，存储扩展容易！分片存储存本地硬盘，不可避免的、由于本地硬盘存不下了，要人肉做数据搬迁或者加一个元数据层进行管理，这不就是GFS么另外，Dremel这个多层树状汇聚，很拉风！！
2021-11-05

峰 👍（3） 💬（0）
好处：不用管存储的高可用，解决struggle的问题。坏处：打破了数据计算在同节点的设计，造成一定网络开销，解决方法：gfs能够提供固定block位置的api。问题：开源OLAP系统中，有像dremel这样可以加入中间层（层数> 1）的OLAP引擎吗? 以及如何确定中间层数。
2021-11-01

LJK 👍（1） 💬（0）
老师好，感觉Dremel的这种计算方式只适合简单计算，如果涉及join操作的话还如何通过这种树形服务拆分呢？
2022-02-19

斜面镜子 Bill 👍（1） 💬（0）
好处理解是本地访问性能和数据质量相对好保证，处理逻辑也相对简单。坏处就是弹性和IO的吞吐会比较限制。当然也想听听作者的解答。
2021-11-01

哈达syn$ 👍（0） 💬（0）
分开存储的好处：计算和存储分离，可以提供资源的利用率，数据量大就单纯增加存储节点，计算量大就增加计算节点，能更好的利用资源。同时任务调度时不用综合考虑节点的性能和数据的位置。坏处：增加了网络传输的时间。这样做是因为硬件性能特别是网络传输设备的提升很大，大数据量的传输已经不是大问题了，数据传输的时间可能比任务等待调度执行的时间还要短
2023-02-19

? 👍（0） 💬（0）
我觉得这个趋势是因为基础设施的发展，最开始数据和代码在一起是因为那时候网络的带宽有限。从远程读取数据对整个系统的性能影响较大。随着网络的发展，网络的开销逐渐不是影响架构的决定性的因素。其他的因素『扩容方便』『容灾恢复更快』占了决定性因素。
2022-09-04

核桃 👍（0） 💬（0）
思考题的那里，问题就是以前存储和计算是不分离的，但是放在了GFS上面，那么数据的容错管理那些就交给了GFS了。但是这里也有一个潜在的问题，数据倾斜问题。不知道多少朋友遇到过。以前使用spark计算的时候，调度算法中如果优先在数据节点计算，那么当该节点中的数据很多都是热数据时，那么就容易出现问题了。当时还出现过生产事故，后面改了调度算法为公平调度才解决的。所以如果避免存储系统的数据倾斜问题，一直以来都是一个痛点和难点，哈希算法目前来说，已经真的快走到头了。
2022-02-22

piboye 👍（0） 💬（0）
clickhouse 做到秒级别
2022-01-16

Jensen 👍（0） 💬（0）
请问老师，Dremel不使用mr进行计算，那么它底层是如何进行计算的呢？
2021-11-28