【重磅】百度分布式计算平台演进历程（从零到全球最大Hadoop/MPI集群）

朱冠胤，08年硕士毕业于北邮，现任百度基础架构部高级技术经理。国内首批hadoop研发工程师之一，带领团队完成百度主流大规模机器学习业务的离线模型训练算法。

2014年8月该团队所参与的深度学习项目获百度最高奖。目前负责领域包括：大规模离线计算、实时计算、机器学习、开放云数据分析等。

link.jpg (36.62 KB, 下载次数: 75)

下载附件  保存到相册

2015-8-7 10:00 上传

典型业务场景这是百度典型的离线计算业务场景，这里MPI面提到的离线计算主要是指计算延迟在5分钟以上的。
按照这个定义，Hadoop、MPI都在这个范畴，事实上百度的大规模离线计算也主要是这两个平台。

百度的大搜索网页建库2009年迁移到了基于Hadoop平台完成，当然在Hadoop平台做了很多改进，来满足高吞吐需求；目前的建库业务已经和2009年相比有了很大变化，这里只说典型的场景，供大家参考。
MapReduce 发展历程
我们先看一下业界分布式技术的早期之路：

• 本世纪初，业界最经典的三篇论文陆续发布，包括GFS、MapReduce、bigtable。
  
• 2004年MapReduce（简称MR）论文发布2006年 2006年，Doug Cutting创立了Hadoop。
  
• 2007年10月份，Hadoop 0.15.1发布。
  
  
  

原因很简单：在MPI平台里面，一次MPI All Reduce操作，就够MR一次job了，多轮迭代的时候，日志重复冷启动、程序和配置都重新冷启动、模型的中间结果也冷启动，多轮迭代的控制逻辑也比较麻烦。

• 百度的两套计算平台合并了，pyramid与Hadoop项目组合并，重点推广haoop；
  
• 由我设计的第一版分布式CTR预估模型正式上线了；
  
• 这是百度第一次引入MPI，而且是在最核心的业务领域。
  
  
  

刚开始的MPI平台采用“手工”调度方式，需要用的时候群里面说一下，很快我们调研了开源调度器，也就是pbs系列。当时选择了torque，也就是pbs pro的社区版。

torque自带简单的调度器功能，上线一段时间以后发现这个简单调度器实在太弱，所以又引入了专职调度的开源软件maui。
2010年的历史结束了，百度也迎来了基础架构部时代。
当然，在这期间Hadoop平台没闲着，继续大规模推广、fix bug，开发新feature，最大的一个feature就是hce（Hadoop c++ extension），在社区还能查到一些影子（最后因为和社区互动过于麻烦，不在坚持推到社区了），感兴趣的话我可以分享完再继续八。
2011年，MR单集群发力2011年，百度MR单集群发展到5000台，这也是业界首次突破5000台规模。
2008~2011，基本都是被业务推着，快速解决个各种规模、性能、稳定性等问题。
2012年，Hadoop2.0全集群上线2012年，百度Hadoop 2.0全集群上线，这个Hadoop 2.0 不同于社区的2.0，因为百度很早就遇到了规模问题，所以比社区提前至少1年就开始研发。不过从图上也可以看出，和社区的2.0架构和原理也基本一致。

2013年继续在规模上发力。
2013年，全球最大Hadoop集群上线全球最大的Hadoop集群上线，单集群达到1.3万台，这也是目前世界上已公开的Hadoop集群中最大的。全集群作业数，达到了百万量级。
在这期间，也继续做了很多资源优化工作，典型的例如存储空间优化。
我们具体解释一下“透明压缩”：

根据hdfs文件热度，在后台将超过一定时间不访问的文件采用lzma等压缩算法压缩。lzma对cpu资源消耗特别多，但压缩比非常好，解压速度也还可以。为减少对业务的影响，压缩的时候利用空闲时段，且业务要访问的时候提前解压。

2013年，我们将百度内部的Hadoop与社区Hadoop进行了性能对比，对比结果如下：


Hadoop的排序部分优化空间较大，这块内存使用也经常出现问题。百度从2009年就将这块的内存采用了类似linux kernel的 buddy system的方式管理，效果非常明显。
这里重点说说shuffle优化，Hadoop默认实现的shuffle问题包括，占用计算槽位。
举个Hadoop实现不好的典型例子：reduce task 启动的时候就将业务streaming方式的reduce程序启动起来了。

这种在绝大部分场景都不适用，这个时候数据都还没就绪，没拍完序就将reduce程序拉起来完全是浪费。shuffle对计算槽位的占用就更明显了，reduce业务根本还没跑，也算作reduce task，而实现又是按照map或reduce调度的。

所以出现平台reduce槽位不够用的情况，因为很多MR作业都在shuffle阶段。百度将shuffle从MR框架中独立成shuffle service独立进程了。所以资源利用率获得了大幅提升。
2014年，DAG执行引擎上线2014年，native c++实现的DAG引擎正式上线，DAG引擎对业务提升非常明显。从如下ppt就可以看出来，错边的图，每朵云就可以看到一个MR作业，MR作业与MR作业之间需要走hdfs写3副本来中转数据。

转成DAG以后，左边的4个MR作业合并成了右边的1个DAG作业。所以很多冗余的磁盘IO、网络IO都优化掉了。还有一种典型的MRr这种业务场景，需要第二个r需要跑一次MR作业，而第二个MR作业的map操作往往是linux下的cat，纯粹是不必要的过一次数据。白白浪费IO和CPU。

DAG效果情况。

继续说效果，上面的图是一真实案例。一个hive作业，默认被翻译成25个MR作业。
2015年，内存流式shuffle上线这一年，更NB的shuffle上线了。很多搞过Hadoop或者分布式系统的都知道，shuffle是最复杂的模块。

简单解释一下什么是shuffle：

就是将源端很多不同类型的数据进过排序、分堆，然后分发到目的端，相同目的端的类型一致。

Hadoop默认实现是map将结果写到本地磁盘，然后通知master节点，然后reduce从master拿到信息以后到map端去拖取。带来的问题很明显，那就是不同的reduce到达时机不一致，因此随机io很严重。
百度新shuffle：

修改为map端完成一定数据量以后向reduce端主动推送，内存流式shuffle。当然这里面需要解决很多技术问题，例如可能reduce会挂，可能遇到慢节点。
2014年的时候，参与该项目的北大实习生同学将demo（名字叫baidusort）拿去参加比赛，获得了2014 soft benchmark世界第一名。
说完了最主要的Hadoop计算平台，下面简单说说另外两个平台。
毫秒级实时计算平台百度内部项目名叫 DStream，能做到毫秒级计算延迟，起步时间比storm开源略早。
大家可能会问，当时为什么没选storm或者s4？这个如果需要，以后可以找时间说明下。

不丢不重的流式计算平台严格保证不丢不重，计算延迟在30s到5分钟级别的业务都适用。因时间有限本次先不讲。

考虑到本次演讲内容较多，因此分为两部分。上文为第一部分，敬请关注第二部分。谢谢。
问答环节Q1： 百度有没有对比与Spark平台的性能？现在流式计算有storm，spark streaming，百度用流式计算的主要业务场景是什么
A1：spark，首先，我们没有把它定位成一个所有领域通吃的计算模型/平台。在百度这种业务场景下，一个模型或者平台如果想通吃所有需求，还能干的很好，我的经验是不可能。所以，我们对spark的定位是：在某些领域使用它。
利用adhoc query ，例如adhoc查询，spark的性能对比，业界也很多了如果是写sql，可能会面临究竟用sparksql还是hive，我的建议是用sparksql，在百度单个DAG作业达到数百万map、reduce的场景下，spark目前还搞不定，业界也没听到单个spark作业太大的例子。
所以，我们对spark的定位是，作为计算平台一个重要模型之一，我们也在重点推广、不过主要是推广sparksql的应用方式，对于流式计算方面，毕竟spark刚开始不是为流式计算设计的，所以在应对百度的业务场景时，会遇到一些问题，我们选择了自研，百度研发dstream、tm的时候，spark的流式处理也还没出来。
Q2：“Taskmanager主要适用的业务场景？如何实现不丢不重？”
A2：适用与计算延迟在30秒到5分钟以内的，当然超过5分钟也可以，不过这个时候可能与MR有一些重叠了。针对这种计算时效性需求，而且要求不丢、不重。典型如计费，如何实现不丢补充？ 目前采用方法是 控制流直接发送，数据流走hdfs。计算模型采用queue worker，上一级与下一级之间通过que解耦，所以非常轻量。
Q3：如何保证内存shuffle可靠性，及时性?会不会有oom的情况发生？或者正在做shuffle的机器意外当机。如何处理这些场景的？
A3：目前做法是map内存流式推送到hdfs，然后reduce从hdfs读取shuffle结果计算，shuffle用的hdfs已经被百度内部提供的分布式内存文件系统接管。本来也有一定ack机制保证，如果map挂了重发会有去重机制。
oom问题，有速度控制，不会出现。或者正在做shuffle的机器意外当机。 map端挂了会重跑map，此时shuffle会去重。reduce挂了的话会从上次ack点重新发送。纠正一下，“shuffler挂了的话会从上次ack点重新发送。”
Q4：百度的计算平台是如何保证安全性的，对任务和用户的管理方面有什么经验可以分享？
A4：数据有owner，获取数据需要申请权限，owner会审批。再说内部平台的话，做好审计就可以了。系统记录下来日志，谁谁什么时间读写了什么数据。遇到问题事后能追查回溯。资源限制方面，下次的架构演进中会重点来讲。没有用kerberos。
Q5：能简单介绍一下百度的网盟推广与大数据平台是如何结合的吗？
A5：我们与网盟有很多深度合作项目 ，Hadoop、MPI、dstream、tm等都在网盟大量使用。网盟的CTR预估也是我们一直以来深度合作的地方。
Q6：针对问题三的，如何控制不oom的？如果shuffler的数据过程中，节点挂掉，如何校验数据到哪里了？
A6：举个例子，要发送前先告诉下游，我要发送10MB数据能接受不，如果下游能收，就告诉map发送端，来吧。这里采用了内存池来处理，如果下游处理不过来会通知上游停止发送。下游会将shuffle结果刷到hdfs上，所以肯定会流动起来。“pendding？是否会导致积压？”，shuffler下游会将结果刷hdfs啊
所以肯定会归还给内存池，然后又可以通知上游继续发送了。shuffler按块向hdfs刷 ，不是所有结果都shuffle完毕了才刷hdfs。
Q7： 百度的计算平台如何做好数据备份的？
A7：很近的数据，在线上服务器本地有一份，然后是线上集群，然后是线下集群 ，当然极端异常了，也会出现一些数据短暂不可用。例如百度的苏州集群就被雷劈过，离线集群，所以集群所在机房可用性没抗住雷劈，当时挂了一批节点，负责运维的同学都知道，有些机器是不能重启的，只要重启肯定就有服务器启动不起来。更何况突然断电这种情况。
当然啊，我刚说了被雷将离线集群批挂过，不过也是几年前了。然后一批机器启动不了，找到多副本都落在这些服务器的”丢失”文件，通知业务方，计算尽量绕过。现场的外包人员尽可能恢复不能自动上线的服务器，同时从其他集群补充丢失的块过来。
有些也可以有一定容忍度，可以接收少量块丢失，也就直接启动计算了。下次也会讲到，百度的Hadoop服务器底层都没有raid卡，采用sata硬盘，所以部分块是找回是需要时间的。所以业务往往就从其他集群补充回来或者忽略这个块了。主节点当然有raid卡，这批服务器质量还是很高的：）namespace有切分啊，这个数据一看就知道是什么数据，可以从哪获得源。然后从源重新下载计算。“你们用的是federation版本？”百度自研的下一代存储，内部名字叫Peta 不过也支持federation。
Q8：那监控Hadoop平台的话，一般会注重哪些参数呢
A8： 很多东西都应该通过dashboard管理起来，例如每天集群作业数吞吐情况、数据量吞吐情况，作业排队情况、map、reduce平均完成时间、排队时间、集群资源利用率，等生成趋势图，如果图的时间粒度足够细，看图就知道是否正常。
Q9：如何实时监控木个job的运行状态，另外对2.0中的yarn调度策略有什么经验分享，用的是那种调度方案？”
A9：某一个job？高优先级打了tag的job么？作为平台方，进来减少太专用的监控。当然实在要搞，自己写个脚本，从他启动方式开始（crontab、监控数据就绪就启动等），最好的方式是给自己保留一定buffer ，确保异常情况下，恢复还来得及。
百度目前的调度系统是自研的，也是下一次分享中将重点介绍的。内部项目名叫Normandy，可以和yarn做个对比，调度策略这块，结合自己业务去调，限制并发啊。 不同队列，不同优先级，可以限制不同的默认并发数 ，防止一次并发太多，其他作业都排队了。
直接限制同时运行的map或reduce任务数就可以了。直接就是解决方案了，例如内存超过1G的，同时运行的只运行1000个。没限制住吧。另外可以限制大资源需求的作业，每台机器最多几个并发。
同时限制最多共计有多少个同时运行，这些策略上去，应该能限制住了。
Q10：不丢不重这块能介绍下关键原理吗，最大的问题是什么
Q10:百度内部这套系统叫TaskManager，采用的模型是Queue-Worker模型，非常适用于复杂的级联业务模块间解耦,控制流直接发送，master节点采用多节点方式，由zk负责主备切换；数据流保存在hdfs上；细节可以看一些ppt中的架构图,它自身设计非常优雅。
Q11: lzma比起lzo，压缩比能提升多少。开源了么，计划开源么
A11:lzma本来就是开源算法，我干的主要是将它无缝集成到Hadoop中了，就和Hadoop直接支持gz一样,现在改名叫 xz了.不过百度用的还是之前的库.
Q12.:冷热数据的分层如何实现
A12:平台本身分成在线和离线集群；离线集群保持大量历史数据；平台根据文件的访问时间判断是否需要压缩；超过一定时间没人访问，当然可以压缩了。以前更早的时候会定期备份到磁带。
Q13: Hadoop什么场景下需要基于hhvm的PHP？另外，据说php 7秒杀hhvm？
A13:百度的Hadoop平台很多人写的MR用php写的,更出乎意料的是还有人用javascript写。集群日均cpu利用率已经在80%了，提升到90%已经不可能,所以只能降低一些不合理的占用；所以率先在Hadoop离线平台尝试hhvm，后来发现效果不错，就全部迁移到hhvm了。再然后就顺便推广到百度所有php场景了。
Q14: 上面提到的百度DAG执行引擎，和spark是什么关系？
A14:百度的DCE（DAG Computing Engine）目的是替换现有MR引擎。将MR作为DAG特化（很多业务看到还是Hadoop MR接口），实际执行是DAG,而spark，我们对他的定位是在某些领域深耕.目前的领域就是adhoc query、图模型等；很多以前用hive的，也都在迁移到SparkSQL。
Q15:百度的优化中哪一点是其他公司最容易照搬利用的？
A15”例如刚说的透明压缩，就这点再展开说一下。很多人的程序只管上线，不管下线。实际这个同学可能升级了新版本，甚至这个业务都已经不在需要了，但没人会将程序下线，继续跑着，所以可能计算结果出来以后，30天都没人用。这种程序是可以通过检查文件访问时间找出来的，然后直接下线好了，预期会节省很多资源。
Q16:Resize机制Hadoop不自带，你们的实现思路是?
A16:Hadoop理想应用场景是每个task处理3~5分钟就结束,在离线集群模式下，不用resize,很简单，需要resize的时候直接销毁重新分配好了
Q17 目前收集日志用的是什么插件？有专人负责么？
A17: 百度内部有几套系统，一套可以用于计费的类pub/sub系统，内部叫 bigpipe，在百度开放云里面即将推出来，叫BQS（Baidu Queue Service），严格保证不丢不重，我们不丢不重的计算引擎就依赖这个传输系统。内部还有另外一套系统叫 minos，用于日志传输，在极端异常情况下，是有可能丢失的，不过在数据量很大的情况下，丢几条也问题不大。还有一些系统trace日志，采用scribe传输。
Q18:百度大数据的生命周期管理是怎样的？降冷如何处理（压缩与存储份数）?
A18:从日志产生就纳入管理，通过pb序列化，导入到数据仓库，主要报表都走数据仓库出了。数据仓库根据数据的价值做出保存周期的判断，给出默认时间周期，如果某些业务需要保存更久，就提供预算过来，压缩、汇聚是肯定的，存储份数方面底层统一做了。
Q19:如何评估业务占用的存储与计算单元（计费）？
A19通过类似账单系统来解决，由业务老大去判断资源占比是否合理，某些小众业务占用太多，基本大家都知道不合理，所以直接就会压制他的需求
不过百度内部通过我下一次将要重点讲的计算架构演进可以获益，通过混布整合，可以让很多资源释放出来，一些没有预算单元的业务，也可以享用一些高质量的资源。
Q20:MPI的存储与计算一般物理上分离，你们应用在什么业务场景较多？
Q20:borg周边配套很多，包括borg config等功能非常强大，borg的新一代调度系统 omega，有时间可以读读，我下次要将的The Google Stack，参考了他的总结，感兴趣的，可以提前bing搜索一下 The Google Stack
Q:21我2009年初基于Hadoop深度优化了业务方写的PLSA算法（一个近点的主题模型机器学习算法），性能比业务方写的提升一个数据量级（毕竟我熟悉Hadoop，可以优化到极致）； 能讲讲这块之所以有数量级的提升是因为优化了哪些吗？
A21:简单说说Hadoop迁移到MPI或者数量级提升的原因。MPI程序可以将输入数据完全cache在内存，数据和字典、配置等保持在内存中，不用像Hadoop每次从磁盘读上来，重新解析,MPI一次reduce操作一个api调用就通过内存+网络完成交换了，而Hadoop就只能多次磁盘IO了,在迭代个100轮的情况下，性能差异太大了。

来源：刘玉强@爱普-北京

审核：萧田国