HCE助MapReduce提升资源利用率

http://articles.csdn.net/plus/view.php?aid=308119

百度分布式高级研发工程师杨栋：HCE助MapReduce提升资源利用率

2011-11-26 09:45 | 3521次阅读 | 【已有0条评论】发表评论

来源：CSDN | 作者：CSDN | 收藏到我的网摘

时至今日，“Big data”（大数据）时代的来临已经毋庸置疑，尤其是在电信、金融等行业，几乎已经到了“数据就是业务本身”的地步。这种趋势已经让很多相信数据之力量的企业做出改变。恰逢此时，为了让更多的人了解和使用分析大数据，CSDN独家承办的大数据技术大会于今日在北京中旅大厦召开。本次大会汇集Hadoop、NoSQL、数据分析与挖掘、数据仓库、商业智能以及开源云计算架构等诸多热点话题。包括百度、淘宝、新浪等业界知名专家与参会者齐聚一堂，共同探讨大数据浪潮下的行业应对法则以及大数据时代的抉择。



百度分布式高级研发工程师杨栋

以下为文字实录：

今天我给大家讲一下MapReduce的框架，这主要是我们自己实现的一个框架，基于C++能够提升资源利用率的一个框架，简称HCE。首先我做一个大概介绍。我做分布式大概有6，7年的时间，刚才刘主编也说过，上Twitter每秒要处理上百万条数据。当然这不是我今天演讲主题，像FaceBook处理每天海量数据，要存储这些数据需要利用到快存储。在传统快存储上要做简单的索引，所以就利用到一些把非结构化转变成结构化的一些工具。在基于这些上面，他要做一些计算，而分布式计算大概分为两个，一类是批量分布式计算，MapReduce是最典型批量计算。有一部分衍生品，像机器学习，有很多迭代性计算，为了高效会有一些延伸。还有现在雅虎说的MPI，是因为MapReduce现象还是计算，而MPI是面向通信，对一些数据挖掘，机器学习方面的东西有不同的模型。

今天我讲HCE，是基于MapReduce的一个扩展。这个HCE框架，本身也是一个开源软件，一年前我们就作为有所涉及。今天我讲的主要分四个部分：Backgroundand Motivation，性能评估，总结。

第一方面我们介绍三部分，现状，遇到的挑战和一个简单解决方案。为什么要提出HCE这个框架呢？首先，我们用集群一般会考虑几方面问题，我们希望集群是容易控制的，这些都是成本问题。第二，我们需要考虑集群的高效性，因为对于百度来说，前阵子一个分会上雅虎公布他最新的数据，雅虎每个月大概有500万个作业，他节点数应该超过5万，每天处理的数据量是超过200PB。百度我们每天有5万个作业，超过总数有1万5千个节点，每天要处理10TB的规模，而且这种规模的扩展是每天都在递进的。作业每天都会增加100、200，数据存储也可能每天以零点几或1%的速度在递增。面对这种不断数据爆炸性的增长，我们需要用合理的手段去控制他，以及我们能够高效利用整个集群的一些资源，而不至于资源浪费。如果能节省一些成本，按现在机器性价比来说，现在一般大存储的机器一般是2到3万，就算定制高性能，大概也是在5万块钱以下。如果你能省10%的成本，就能为公司节省上千万的财产。所以说，技术上的一个小缺口，可能给公司带来很大的收益。

HCE目标有两个方面，第一，需要提升整个集群使用效率；第二，附加价值给开发者一个更好的开发接口。面对的挑战，提升集群资源利用效率，业界一般有几种做法，最典型是做以兆级优化，一个是资源调度。换句话说，主要是在资源调度上有大的改进，传统调度是每次作业就是两阶段来提交，其实MapReduce不能准确进行描述，还有存储等等。所以，新的MapReduce2.0会把这些资源调度描述更细，他通过分配的一些算法，一些资源分配来描述每一个作业真正需求，真正能够把作业力度切到更细，这样整个集群资源使用率也会提高。

第二种方法是Task optimization。因为用户提交一个作业，集群是不知道用户什么配置。大家都了解，一个作业可能需要上百个配置，而这些配置用户怎么能正确配置好，而且很多用户他是不愿意去配置的，即使是在公司内。如果在平台方看来就需要做一个动态，这块是比较难做的。我刚才说的这些都是业界传统一些优化方法，当然这些传统方法我们也做了。我们有一些另辟蹊径，每一个作业被切成多个来运行，实际上作业执行效率，除了调度期快慢以外，还取决于单机执行。

而Task的优化很多人是不去关注的，我们在这里分了一下Task，什么叫small tasks？什么是big tasks？如果用户作业是零就由框架来占，什么是用户开销是零呢，如果用户没有做任何结算，你数据流也要流过整个框架，管这种作业叫small tasks，用户就没有其他额外的工作，这就是一个small tasks。而做一些工作你会发现这一个small tasks会跑的很快，会在几十秒内结束。对用户程度比较重，用户需要做迭代，需要做复杂的的操作，这个任务会执行很长，会几分钟，甚至几十分钟。

总而言之，我们把那种用户，用户定义的很简单执行时间很短的任务我们称为small tasks，对于哪些用户在营销里面进行大量的，大家都知道这是一个主要因素，我就先指出这里面有复杂计算逻辑而且执行很长时间的我们就叫big tasks。

对集群做一个统计，因为你抽象为要抽象small tasks和big tasks，我抽取四个主要因群来看的话做一个统计，统计的标准是什么呢？这个tasks到底执行多长时间，不是一个作业，而是一个tasks，tasks是作业切分之后，平行化之后在电极上执行的任务叫tasks。通过4个集群统计会看到，超过80%作业，其实他本身任务都是小于2分钟就完成的，而且应该是超过75%都是在一分钟之内完成。

因为本身这是符合逻辑，你的集群跑到一个公有计算平台，一定是轻量级。大家都了解机器学习里面有一些复杂的迭代性算法，相关有一些MPI五做的这些东西是处理那些复杂计算逻辑。一般统计无非是操作，都是很轻量级这是符合逻辑。比如任务都是小于1分钟，而且你要跑Map的时候就需要知道，我们统计得出Map数据基本上是Reduce数据量倍，这样数据量会大幅下降。一般根据FaceBook统计结果，平均有5到10倍介绍，你输入假设100倍，收入就是10个，Reduce是很轻量的，你能大幅提升Map，因为你的大头在Map，这一个发现。

换句话说，本身对于那些轻量级的。第二点对于用户来讲，在国内大部分人使用写CDI出身，让你能够实现多语言支持，你可以写C++，可以写各种脚本，说白通过管道，通过pad通过STD，把数据传给JAVA进行执行。这样的话有一个问题，用户作业是独立于你框架，你框架只干了一件事就是把用户任务启动起来让他执行，看整个框架是怎么编译优化。有人说我框架不需要关注这个东西，因为很多用户，比如说用户在编制程序的时候不去加JAVA，或者用户用低版本，本身学过编译的人都了解，编译的好坏，编译手段不一样，也会产生影响。

所以，在我看来我们需要找到一种手段，首先第一我能优化我的方案，第二优化用户作业。前面说过比较难实现就是如果从配置上讲，最好能配置资源，当然动态配置是因为你需要了解，思考，去模拟用户作业到底是怎么执行的才能去做，而我们这里提出一个比较确切，比较简单方式，我不需要了解这些配置，我从tasks静态解决这些问题。

我们看一下Hadoop MapReduce，来看一下Task Runtime，可以看到是通过Task的JVIM，这只是一个框架，这里讲的是去负责启动用户，用户进城之后通过管道，标准输入，标准输出来做，这样的话其实很难做到对用户进程的控制。为什么要用JAVA呢？这是Hadoop最早提出来JAVA需要跨平台，做系统软件你知道用JAVA实现跨平台很难，对于现在国内互联网公司来说，所有服务器，以及系统都是定制的，他不需要跨平台的功能。

第二个因为JAVA编程语言特性大家都知道，Hadoop为了更好的收缩性，他分装了很多层，在编程者看来这是有必要进行简化的。第三个最重要就是JNI，Hadoop是怎么做的？压缩都是本地库，压缩方法有很多，比如Linux自带的，包括LGO，包括Google最近开源压缩库都是C++本地，只能通过JNI，因为JNI比传统高效一点，所以在Hadoop里面是通过JNI来做的。

第四点你本身的程序是独立于框架的，所以你框架是很难做影响的。这一页就给大家介绍一下现有实现的四个问题，我们怎么来解决。这是我们的目标，我们总目标基于原来框架，我们目标有两点，第一就是提升整个集群的资源利用率，第二就是附带解决开发效率的问题。提升资源利用率，如果我们想象把整个框架和用户，这幅图，如果整个框架和应用绑定起来，和用户能绑定起来，我就能做最简单的编译优化，用最简单的方式去进行实现。

第二，我们希望用C++来实现，为什么？因为我是做Hadoop出身的，所以我用Hadoop创始人那一句话有两点，因为你的MapReduce是C++密集型计算，当然你这里不是内容密集型，因为每一个tasks执行都会推出，但如果在2.0，你的tasks为了更高效不推出去接新的tasks，对于C++来说，JAVA用的久人也知道，最近JAVA最新JNI优化器优化之后，比C++性能差不了太多，原来大家认为差5倍，4倍，现在差2，3倍，当然差距还是有。选择C++的话，一方面比JAVA效能高，另外相对全面应用也有好处，第三点JAVA是虚拟机自己控制，而C++里面可以更快释放。所以，对于资源管理更好，当然这些不是我的观点。

前面回顾一下我介绍的背景，因为我们的集群在不断的扩张，而且我们集群的使用效率会有一些问题，基于这种现状，我们除了在作业召集的优化，我们还做了tasks级别的优化，我们是要解决两个问题。第一要提升资源效率，框架的资源效率，第二我们希望把框架和应用程序绑定起来，这样在优化框架同时能优化应用程序，静态的方式。具体该怎么做，我们介绍一下。MFramework Model，首先看一下整个框架中位置，第二看一下功能模型什么样子，第三实际处理模型什么样子，第四看一下接口提供哪些编程接口，最后对比一下HCE和现有的Hadoop。

Overviwe，整个HCE的位置就如红字所设，是相当于一个处理引擎。说白了实现了Hadoop里面非调度那层，因为在Hadoop里面调度是有JAVA做的，执行是有C++做的，因为调度没必要做成C++，非内存密集型。所以，你只需要关注，这样你可以把Hadoop分成两层，第一层是调度，第二层是调度，执行用C++执行，调度用JAVA。

因为附带会提供一些其他的编程语言接口，因为传统现在一些编程语言接口用的比较多，Hadoop作为用户一般是用JAVA，还有以数据仓库来做，通过多变，HCE本身是C++实现的，他最终提供是C++接口。C++大家都知道，支持其他编程语言接口非常方便。因为其他脚本语言，所有都是基于C实现的，换句话你只需要做一个最简单解释器就能实现对接，而且更高效。

他底层是基于HDCSS，尤其在国内很多系统都是用C++实现的。如果用HadoopC++实现数据库的话，C++框架直接去接C++的数据库会更高效，不然的话你JAVA想把MapReduce加在C++实现的框架上，必须要实现语言转换。所以，他对底层支持也会更好一些。整个数据通路，一般传统数据分析流程是前端WAP抓到数据之后，给到最前端的中间件，或者KV，Data，他们通过一些云传输，像开源，或者通过一些大块传输传给后台的分析系统，分析系统做一些计算，或者做一些MPI计算，这样有上层JAVA进行读取相关数据分析，反馈用户最经典就是Twitter的方式，通过简单实施流处理。

因为实施流处理不能脱离，因为有窗口概念，最终结果可能还是需要P处理来校验，这方面不多说。我们还是回到MapReduce本身，本身我们一下具体该怎么做。MapReduce阶段本身的Map阶段，首先他是有读文件，一个任务会把文件读进来，第二是做一些Map处理，相当于是去引用用户的一些方法，因为在一个Map处理里面，你用户计算的这且东西越来越大内存可能放不下。因为你每台机器可能跑8个Map，16个Map，内存放不下的时候要设置起来，有一个设置阶段。还有我先写一个临时目录，最后再提交到正式目录这么一个过程。Reduce有一些Shuffling为什么重点关注Map没有重点关注Reduce，那会我讲的时候第三页，我说过，有一个图，本身Reduce个数就比Map要少很多，一般轻量级计算少5-10倍很正常，而且Reduce的计算是很轻量级的。

所以，你用框架去优化就够了，不用对Reducetasks做额外优化。而且Reduce很大一部分是在Shuffling因为你是并行计算，多个Reduce需要知道输出结果，就需要有Shuffling，多点到多点的数据传输，在Web2.0需要其他方式实现，我通过实现Shuffling脱离，把这部分单独剥离出去，相当于有三个阶段，总的来说还是有两个阶段，Shuffling相当于有一个独立的服务，当Reduce输出外之后会通知Shuffling进程，相当于服务托管，说我算完了，Shuffling服务把他把Reduce输出结果拖到Reduce阶段，传统Reduce是要站槽位，占资源那一刹那并不能计算，把Shuffling脱离出去之后就不用占用这部分资源。

所以，HCE不解决这个问题，HCE只解决Map问题，而Shuffling的问题要留在2.0解决。看这个图一个首先会通过Map会产生多个文件，之后会进行排序，做一些设置，最后合成整个输出，把这个文件再Shuffling到其他Reduce处理节点上进行处理。HCE其实主要是解决Map端，因为Map端后面测试我们可以看到，Map端实际上占性能开销最大还是OI相关的，以及比较占CPU资源的槽，我们就需要进行独立解决。

本身功能框架大概分四个层次，最上层提供了Java，C++，Streaming，Python，Php，相比于原来只提供Java接口，他认为如果你是原生态就做Java，其他语言统一Streaming，这样问题开发者会有一些变动，第二Streaming还是有一些管道开销。处理每一个KV都需要管道，管道就是拷贝一次，都会额外有两次拷贝到Java里面，如果省去这一点可以节省一点，有人说节省这一点重要吗？对于一个上万台集群，只要节省1%就赚了几百万，节这么一个概念。

第二层次是Compiler Optimization，对编译做的。还有一个代码实现，能够从C++转到上面的Php接口，这个其实代码量100内搞定，很方便实现多语言。执行层没什么好说的，执行层和Hadoop原来都是一样的，只不过我们原来做了一些优化，让每一个模块都比原来高高兴一点。底层的压缩库，我们做了一些调优，因为大家都知道，发明一个压缩算法很难，因为像传统是有几十种压缩算法，我们只需要针对不同数据去选择不同的压缩算法。还有存储接口，可以和C++存储更好交互，换句话能够在其他C++实现的上节省很多IO如果你要通过GNI，其他的方式来做。

文件格式传统Hadoop支持两种，MapReduce，文本等格式。看一下整个数据流，整个HCE数据流，用户提交作业从这端开始提交，到切割处理，相当于上层只是在Java只是一个虚拟的代理，真正实现都是在C++上面。其实早期做了一项，通过C++空间来实现Shuffling，其实效果不大，本身瓶颈不在于传输，而在于本身在2.0里面，把部分槽位省下来，本身性能有很大提高。其实C++端贡献最大还是Reduce这块，相当数据流这端，最终数据输出在控制上层做完就OK了。

换句话说，把数据切到C++的第三个，为什么还要实现，因为很多作业已经用Streaming跑，我们增加了StreamingOver Hce的接口。这里SSE怎么利用静态编译去优化一个程序？实现一些传统方法向，这种操作对每一种都有3到5倍，甚至10倍提升。有人说这些用户他可能不用，换句话说SSE就是强迫用户用，用户用C编程必须要去包HCE框架，这种SSE指令级带来性能提升。

最终会提供这么几个接口，像C++，CHE等一些接口，性能大概有10%-30%提升，而谁在用呢？像Java是Hive在用。对于一些需要提升性能的，因为是，还有一些比较大的用户程序占了很多东西作业，他是用C++就能有很多提升。进行一个对比，HCE对比Hadoop有这么几个方面，提供编程接口更多一些，很容易支持其他存储系统。第三，本身要比基于JNI的性能提升很多，再就是我们用实现静态编译，使用户进程能够跑的很快，当然会做一些像比较久远算法，在大数据里面，因为默认Hadoop最终容易实现combiner，这是什么呢？就是Map阶段去做Reduce，这是很重要优化。所以，HCE用这些技术都会比Hadoop更优。

最后我们看一下对比，他的性能比，这是原来Hadoop，什么叫Timings，我把所有Shuffling切成一段段对比，HCE取两块，其他都是一些功能型实现。为什么优化呢？因为第一C++在本地做排序，第二我们有用JNG，我们考虑 压缩因素，压缩算法有很多，尤其是中间结果压缩，基于本地的，换句话说你在本地，大家看这幅图，什么情况最高呢？他恰恰中间结果用了IOGO压缩最多，换句话这个压缩比最好，耗的CPU最多，需不需要用这个呢？不一定，大家去看官网，其实你做压缩，本地本身跑计算这些数据已经是半结构化的数据，或者结构化的数据，他做到这一部，用其他压缩法压缩比也不会差一倍，本身CPU消耗，包括Google等等这些东西也比哪些高很多，这是10个节点100G测算结果。

如果我用SSE指令来编译的时候，利用编译优化的算法还有额外10%的提升。第二个应用，这是百度实际应用，跑两个实际应用，第一个是语言的影响，第一个是Hadoop传统Streaming，大概跑了50秒。本身HCE，这是差不多在90台机器上跑的，用HCEStreaming有所提升，你省去Streaming管道还有一个提升，传统跑了20多秒，变成Streaming跑了这么多等等都不同，所以根据实际来看优化框架和静态编译优化程序我们都做到了。

最后总结一下，不应该叫Jobs，我们如何去优化一个tasks，我们HCE目标就是优化tasks。首先你要通告combiner，保证在Map端数据减少，到Reduce就很轻量级。第二是用C++接口，第二通过压缩算法等来进行。Contribution大概在今年年底，所有集成作业都会切到HCE上，当然是百度的。第二就是Applications，有哪些用户，有哪些作业？他任务都很大，很重量级。第二就是MapReduce-based warehouse，这就是我那会说的话，HCE本身会节省超过10%机器，为公司能节省，如果全部用上的话能节省10%。

什么叫Hive Over HCE。有一个同学在Hive工作，我把HCE推荐给Hive，他做了一些试用，以及他们跑的一些作业。FaceBook这边做了一些简单实践，他相当于把MapReduce，因为大家了解Hive就是MapReduce一层分装，把Hive和Reduce本身实现复杂逻辑。Hive支持是劣存储，他实现这些东西，然后就迁过去。他给出数据，他们实际跑的FaceBook一些作业有20%到50%性能提升，为什么平均30%呢，不是很高呢，因为他的作业都是很重量级。换句话FaceBook作业都是好CPU，什么在好CPU，超过70%是压缩和解压缩，为什么？因为国内这方面可能做的不好，国外所有的输入输出都是利用Jira压缩，现在国内都是计算式瓶颈，有的人觉得不差钱加机器扩容解决，国外这块做的比较精细一点。

有没有办法解决呢？我建议一种方法，用SSE把压缩库重新编译成内联的方式。因为所有的压缩都是用我前面说的这些简单的语言，函数实现，而压缩库本身是是0的，你必须利用高效指令进行优化，因为Hadoop利用到这种技术。所以，这是一个额外话题，你希望去优化程序，你应该去关注程序本身性能损耗在哪里，有没有相似或者有没有一些简单不用去实现那么复杂的动态配置来解决这个问题。