1006大设计翻译版

本我，本及其研究工作是由在导师指导下独立完成的，在完成时所利用的一切资料均已考文献中列出。：SCOPE，性能分析，大数据任JobPerformanceStudyonBigData WeiLiWithwide-adoptionofbig-dataplatform,moreandmorepeoplearecapableofrunningtheirdata-parallelcomputingjobstoyzethemassivedata.Suchjobsaredividedintomul-tiplelogicalstages,andeachstagewillspawnhundredsoreventhousandsofparalleltaskswithsignificantdatathroughput.Thereforejobperformanceiscriticalinsavingbothtimeandresources.Theperformanceresearchonbig-domputingjobshelpsustolocatepoorper-formedjobs,findoutthedeficienciesofexistingbig-dataplatform,andfurtherguidethefuturedata-parallelprogramming.Asthefirststepinthewholeresearchwork,weconductedanempiricaljobperformancestudyinSCOPE,aproductionbig-dataplatformin.Thisstudystartswithtwokeyfactors:degreeofparallelismanddatathroughputtodiscovertherootcauseofthecommonperformanceissues.Afterhavinginvestigated124realSCOPEjobsthoroughly,wefindthattherootcausesforpoorperformancecomefromthreeaspects:programminglanguage,dataandsystem.Wererunsomepoorperformancecaseswithoursolutions,savingupto64%Thebig-dataprogramminglanguagemakesusersfocusontheirbusinesslogic,insteadofthedetailsofparallelism.Howeversuchconveniencesometimeshindersthejoboptimizertoproducebettercode;Thedataimbalancemayleadtosignificantfluctuationsonexecutiontimeoftheparalleltaskswithinastage;Thestaticjobexecutionplanmaydecidetoolowortoohighdegreeofparallelism,whicharealsothereasonsofpoorperformance.AlthoughourstudyisbasedonSCOPE,themethodologyandfindingareprac-ticabletoothersimilarplatforms,includingthreepoorperformanceissues.Andtheresearchre-sultswillbeguidancetothedesignandoptimizationofbothbig-dataplatformanddata-parallel:SCOPE, ysis,BigData1简介11.1课题来源11.2课题背景11.3课题意义21.4国内外研究现状21.5研究内容和目标31.6研究难点与结果41.7组织结构6272.1MapReduce编模型72.2SCOPE系统72.3SCOPE概念与实例82.4数据倾斜（DataSkew）现象 3方法论 3.1任务的代表性 3.2研究的适用性 3.3人工判断原因的性 4SCOPE任务性能表现具体实例分析 4.1语言层面 4.1.1现象与危害 4.1.2实例分析 4.1.3解决方法 4.1.4总结分析 4.2数据层面 4.2.1现象和危害 4.2.2实例分析 4.2.3解决方法 4.2.4总结与分析 4.3系统层面 4.3.1现象与危害 4.3.2实例一分析 4.3.3实例一解决方法 4.3.4实例二分析 4.3.5实例二解决方法 4.3.6总结与分析 5相关工作 6结论 致谢 参考文献 本课题来源于航空航天大学软件开发环境国家和微软亚洲系统组的合作项目，关于微软SCOPE大数据平台性能研究和优化。课题的整体方向是在了解SCOPE系统，了解SCOPE任务的整个调度过执行过和优化技术的前提下，从当前的SCOPE大数据任务的执行情况和系统日志出发，SCOPE和其他大课题研究中涉及到的所有数据均来自微软SCOPE产品组集群，覆盖搜索、、数据环境下失效，和计算重大的。海量数据的现实对于分布式计算的编MapReduce类型的数据并行（data-parallel）计算模型是将一个计算任务分成多个不同的计算阶段，而每个阶段包含多个可以高度并行的子计算任务，这些计算任务针对不同的数据进行相同的计算，而不同的阶段之间有的需要一个称之为shuffle的阶段进行数据的清理与重新组织。在整个过中，不同的阶段都可以让用户定义数据进一步的，类似SQL的次的型语言和过型语言（e.g.C#,Java）编写的用户定义函数（UserDefineFunction,UDF）结合的编模型已经在当前的大数据平台上成为主流。这些平台包括雅虎的PigLatin[2,3]，的Hive[4,5]，的FlumeJava[6]MapReduce过在底层的集SCOPE（StructuredComputationsOptimizedforParallelExecution）[7据平台，它拥有类似SQL形式的查询语法并且支持用C#编写的丰富的用户定义函数，MapReduceDryad[8]，能够很好的扩展和容错。SCOPE目前运行在成千上万台机器上，服务于必应搜索、数据挖掘、等众多产品部门，每天运行着数以十万计的大数据任务，处理PB级别的数据。数据平台任务的一次运行通常会涉及多台，有的时候多达数百台甚至的机器，而任务性能的提升可以帮助系统节省计算资源和能源，进而提升系统的计算能力，从而可以让集群支持的计算任务。而从方法上看，大数据平台上的任务性能分析，与传统单机序和多线并行序的性能分析有很大不同。大数据系统相比于其他系统更加的复杂，因为容错和综合调度等复杂机制的存在，其执行过更加的多变，而造能不佳的原因也与传统任来指导这样的新环境下的性能分析工作，而这方面的相关研究目前非常的少。对于大数据平台的用户来说，性能研究也是有很大意义的。大数据平台将用户表达的次的数据处理逻辑转化成分布式并行计算的任务，调度安排到成千上万的机也因为这样，用户序和最后的执行代码存在较大的差异，从而导致用户对自身序解任务低性的根本因，从指导用在数据平台上编实，优化统，节省资源。大数据平台上，对于产品部门的大数据任务的性能研究，能帮助了解真实生产环境中对于大数据平台的应用情况和实际的需求，从而发现当前大数据平台存在的根本问题，并未来的系统研究和优化方向。MapReduce提出后，在工业界得到了广泛的追捧和应用。Hadoop是最流行的MapReduce实现，它已经逐渐发育成为一个成开源生态系统。目前，基于MapReduce计算模型，针对不同的应用类型，越来越多的应用框架都可以在Hadoop体系中部署，包括实时计算的Storm、Impala，内存计算的Spark，数据平台Hive、PigLatin、HBase，DAG图计算引擎Tez等。越来越多的公司在他们的实际生产应用中部署平集数和分一今似SL的次型语言和过型语言相结合的编的方法已经成为上层大数据任务编的流行趋势和发展pdueSOPE平台也是这样的于Hive和Pigatin，它提供海量的数据和计算服PB级别的数据。性能分析的研究方面，之前关于大数据平台的性能研究工作多针对研究型集群[9]，本课题的研究目标是：了解当前大数据平台上任务的运行情况，分类找出不正常任务的效率。我们的研究将对于真实应用场景下的任务的运行表现出发，分析具体影而总结现今的大数据平台下编实践，更好的指导今后的分布式系统设计。熟悉和了解微软的SCOPE大数据平台，包括平台架构、编语言、调度机制、原始数据，从SCOPE大数据平台上真实的大数据任务的原始数据及相 大数据平台上的任务性能研究有其复杂性。现在大数据平台上运行的任务有着复杂的计算逻辑，再加上系统将用户代码并行化的过中应用了众多的优化和调整，最终导致了任务的低性能瓶颈很难定位。而造成任务性能不佳的原因也有很多，包括用能，甚至连同一台机器上运行的不同任务的tak之间也可能会有相互的影响。这些都使得任务性能分析和研究充满了与。与传统的序不同，大数据平台上的任务旨在通过并行计算的方式快速的进行海量数据的处理，其任务性能取决于并行的任务数和并行执行的效率，而不是传统的复杂度最大的模块。用户提交任务到大数据平台上，均希望能利用其分布式并行计算的优势，能够方便的和处理海量规模数据，更快更好的完成相应的计算任务。在大数据平台中，任务运行的时候不可能的使用资源，所以系统通常会用某行的任务数来保证该任务的计算资源，在SCOPE里称之为额定并行度（TokenAllocation同导致其需要的并行任务数量一般是不一样的，而现在的大数据任务往往都比较复杂，不只是简单的一个MapReduce过而不同的阶段中又有一些是可以并行执行的，从而执行的任务数是由用户序逻辑、数据和系统等多个因素共同决定的，有可能存在某段时间远远达不到系统给定的额定并行任务数，也有可能某段时间远大于这个数值。本文认为一个好的大数据并行任务应该充分利用系统给予的并行计算资源，保持整个过较以并行执行的子任务数称为运行时并行度（RuntimeDegreeofParallelism,RDoP，作为吐量（Throughput）作为重要的参考指标。RDoP可以在宏观层面上反映一个任务对于额ThroughputRDoP的不足，帮助研究每一个并MapReduce这样的为海量数据处理设计的系统的假定，从而可能给系统带来额外的开SOE（DoP和子任务吞吐量（Throughput）这两个指标，展开了第一个针对产品组大数据平台的经验性的性能分析研究工作。我们分类得到了一个潜在的低性能任务集合，并对其中的124个不同的产品任，进行了工的深入究。本文的研究现，这些务低性能本原因主要可以归结为语言、数据和系统三个层面。本文对于这三个方面的原因进行了深入的分析，并总结了当前大数据平台研究和性能优化所的和。在大数据平台的语言层面，因为大数据平台普遍采用比较容易编并符合数据处的SQlike语言，同时还支持过型言来灵活实现用户定义函数（DF）,这户编充灵性低写算序槛用不理繁杂的并行细节。但是这样的代码对于系统底层的优化器来说却并不是那么的友好，特别是在海量机器参与的复杂集群中，上层序的过于灵活导致了有些重要的优化难以Prtialggrgte这样一个显著减小hufle的，个有行Prtialggrgte优化，包括那该更慎重的考虑编语言的设计，编语言不仅要能够让用户更好表达自己的处理逻辑，而且对于系统优化器也应该是友好的，这样才能做到编灵活性和效率上的兼顾。在数据层面，大数据平台上采用数据分块的方式来方便同一个阶段内任务的并行执行，这潜在的对于数据分布的均匀性带来了要求，数据分布的不均匀会直接导致同一阶段内不同并行任务间的运行时长不同，这将延长整个阶段的完成时间，从而带来低性能的问题。而且目前大数据任务应用的广泛性和任务的复杂性又导致了任务执行的中存多个阶段不仅仅原始数据，中数据也要是均匀，这无是一巨大的。大数据平台上的系统实现也存在很多的。用户任务在大数据平台上运行不只是追求任务运行的正确性，更加看重任务的效率与性能，这使得任务的并行度变得尤为的重要。而前系统中灵活的编语言复杂的用户逻，均导致静态的并务数决策在现在的应用环境下不足以满足任务的需求。一个计算阶段中，如果并行度如果一个阶段中，并行度过高，会直接导致计算任务中每个子计算任务都只是处理极小的数据，这使得整个算过系都有贵的间接开销性能不好于是，在统实现的角度上，对于不同阶段的并行任务数进行动态决策是必要的，但也是有难度和的。产品组集群的任务特性使得可以结合产品任务中任务反复运行的特点，根据历史任务的信息，结合运行时动态的数据大小监测，采取更好的决策策略。本文在进行任务运行缓慢性能不佳的根本原因分析的同时，提出了一些解决当前任务性能缓慢的方法。我们对于部分仍保留原始数据的任务进行了重新运行，我们提64%的任务运行时间，并且减少了任务因为运行缓慢而最终失败的概率，有效提升了性能。随着大数据平台的应用变得更加的广泛，在这个新兴的领域上的存在很多的问题等待我们去思考，任务性能的分析是我们研究的第一步，帮助了解系统中存在的一些现象，•两个衡量并行任务性能的重要指标，并利用该指标制定了3个有效的方法组织结构如下第二章介绍相关的技术，SCOPE的相关知识，方便后面具体描述并行计算任务的MapReduce是公司，在廉价机器集群上做海量数据计算的分布式系统。MapReduce包括Map和Reduce两个过Map过进行过滤和的分组操作，Reduce的过进行一个聚合的操作。Map和Reduce这两个过都采用过型的语shuffle的阶段，进行数据的清理和重新组织，系统将按照用户指定的Reduce阶段的关键字，将pdue点上的子任务出错失败时，会通过系统自动的按照出错的类型进行重新运行，而不需要重新运行整个任务，也不需要人工介入，这有效的提升了任务的执行效率了在超大规模并行计算任务得以顺利的完成。MapReduce编模型具有非常容易扩展的特点，结合系统中拥有复杂的机制来保证整系的负载均衡、异常检测、失败恢复等等，很好的满足了海量数据处理的需MapReduce模型才能在这SCOPESCOPE是微软开发的应用在微软的海量数据处理系统，其底层是类似于MapReduce的Dryad分布式计算框架，而上层支持用户采用类似于SQL的型语言或者采用CHivePig。SQL天然适合进行数据处理的操作，已经获得了广泛的应用，具有简洁的编方式和丰富的语义表达这样优秀的特性，可以提高用户编的效率；系统还支持过型语言编写用户定义函数（UDF，使得SCOPE语言不失灵活性，能够应用在的大数据处理场景。SOPE序执行的各个阶段中，一般把数据作为结构化数据集合来处理，每个数据行包括多个列，数据集就像数据库的一张表格一样。数据集是实际代码中每个语句操作的数据的基本单位。在实际系统实现上，SCOPEMap-Reduce-Merge的模型，Merge是指将两个数据集合并为一个数据集的过在SCOPE中，一般把MapReduce中的Map阶段称为PROCESS阶段，Reduce阶段仍然称为REDUCEMerge阶段称为COMBINE阶段，为了方便，本文后面都采用SCOPE的说法。SCOPE系统因为其上层提供简洁高效的表达方式，下层有着强大的海量数据处理能力，在微软的很多产品组中得到广泛使用，用来编写处理海量数据的分布式并行计算任务。SCOPE很容易编译成类似多个MapReduce序互相连接的数据并行（Jobt0=SELECTt0=SELECTaasbasBFROMt1=SELECTaasSUM(b)asB,SUM(c)asFROMrawtable1;t2=SELECTt0.*,FROMt0,t1WHEREt0.A==123456789这段类SQL的代码中，t0rawtable0这个数据表中ab两列，分别称为AB，t1rawtable1这个数据表中选a,bc三列，并且分别bc两列进行一个求和操作，把生成的数据集合三列分别称为A，B和C，接下来，t0和t1这两个新的数据集合按照A这一列进行接操作，这里把A称为joinkey，得到一个新的数据集合t2。如代码中所示，SCOPE支持类似SQL的语法持常见的多种聚说COUNT,COUNTIFMINMAXSUMAVGSTDEVVARFIRSTLAST等，它们的图 运行情况COUNTDISTINCT就是一个常用的计算某一个数据列中不同元素数量方面的语法。SCOPE还支持相等连接操作，也就是SQLJOIN操作。这段代码被编译成类似图2.1所示的样式，包括PROCESS，REDUCECOMBINE三个阶段，每个阶段中都包段将不同的Vertexprocessor、reducercombiner。它们最终都将通过调度器调度，从而并行的执行。processor一般是将数据进行格式转换和过滤，其输出的某些行按照用户指定将被当做reducekey，在经过一个数据处理的重新分堆处理的过后，reducerreducerSQL语句表达，reducer计算内容由用户自己定义。processreducedatashuffling的阶段processor中的reducerkeyreducer的机器上，而reducer机器上将对于这些数据进行一个多路归并的操作，而后再执行reduce过Datashuffling过会给网络和系统带来很大的负担，因为所有的数据都将通过网络进行传输。在实际实现中，为了减少datashuffling的过的数据量，对于那些reducer的计算操作是一个聚合操作，而且是满换律和结合律的聚合操作的datashuffling开始前先在本地进行相同的聚合操作，这个优化称为PartialAggregate。在实践中，很多的任务都包含聚合操作，所以这个优化能有效提升datashuffling的效率。对于复杂的SCOPE任务，一个MapReduce或者Map-Reduce-Merge过可能没有办法完成。最后，SCOPE代码会被编译成为一个DAG图进行执行，DAG图上的每一个节点称为一个Stage，一个Stage可能对应为一个PROCESS、REDUCE或者COMBINE义、用户定义函数和系统的一些优化技术，每个Stage中包含多个Vertices，每个Vertex包含整个Stage处理的数据流处理其中的某一部分数据。SCOPE调度系统会按照这个DAG图的依赖关系，调度并行执行对应的Vertices节点。SCOPESELECTSQLC#写的自定义处理函数或者聚合函数，用户还可以写MapReduce形式的序。通过自定义的PROCESS、REDUCE和CONBINE来定义更加灵活的操作。SQL操作也可以用如下的方式重写。这里的MyProcessor、MyReducerbinerUDFSQL的语义，可以灵活的表达自己对应的业务逻辑内容。要达到与上面的SQL-like代码相同的语义，MyProcessor和t0=PROCESSPRODUCEA,t1=t0=PROCESSPRODUCEA,t1=REDUCEONPRODUCEA,B,USINGt2=COMBINEt0WITHt1ONt0.A==t1.APRODUCEA,B0,B1,123456789IEnumerable<Row>reduce(IEnumberable<Row>input,Rowoutput){intsum_b=0;intsum_c=0;foreach(Rowrowininput){output[0].Set(row[0].String);sum_b+=row[1];sum_c+=}output[1].Set(sum_b);output[2].Set(sum_c);yieldreturnoutput;IEnumerable<Row>reduce(IEnumberable<Row>input,Rowoutput){intsum_b=0;intsum_c=0;foreach(Rowrowininput){output[0].Set(row[0].String);sum_b+=row[1];sum_c+=}output[1].Set(sum_b);output[2].Set(sum_c);yieldreturnoutput;}123456789DtaSkw行计算中，很多时候不同阶段需要一个等待同步的关系，等待一个阶段中所有的计算任务完成才能进行下一个阶段，这个时候计算阶段的完成时间就取决于计算任务中耗以DtaSkw现象的出现在很大度降低并行计算的效率。现象的严重性并没有得到应有的认识。在后面详细的进行解释。SCOPE产品组集群是一个典型的大数据计算任务平台，每天有上万个的任1000个任务作为本文研究的原始数据集合，这些任务中很大一我们的研究发现任务运行时并行度（DoP）rtex吞吐量（Throughput）是大数据平台量数据规模下的并行计算任务运行表现中最重要的特征，可以帮助我们时刻运行的并行节点数越高反映了该时间内任务对于并行资源的利用越好，这样的时间越长对于一个数据并行任务来说越有利，因为它的使用到并行资源进行任务加速。吞吐量则是一个微观的指标，可以描述并行计算的子任务的性能表现。在这样的并行计算平台上，调度一个子任务进行计算需要一些调度时间和启动时间，那么如果rtex该rtex，进行有效计算的时间比例不高，如一个任务中绝大部分的计算都存在这样的情况，那么这个任务很有可能性能不佳。为了更加有针对性的分析任务性能低下的原因，我们结合上面提到的两个重要的衡量任务表现的指标因素，制定了下列几个规则来帮助分类找出这样的潜在的性能低下的任务。这里的“连续的”使得我们找到的长时间低并行的任务执行过是在任务执行的一20%是一个经验值，我们认为如果不正常的低并行阶段已经占到任务总时间的五TokenAllocationJob可以有足够多的额定并行Job1Job表现为有一个或者几个Vertex运行相当长的时间，而这个时候已经没有其他的子任务可以与其并行执几个花费异常多时间的VerticesStageJob的完成2JobStageVertex都开始了，但是很长一段时间都没有任何Vertex完成，这显然也是不正常的，很有可能是因为每个Vertex的计算复杂度过大或者处理的数据过多，但是即便如此，这个时候Job的并行度还是低于额定值的一半，意味着仍然可以通过调整Stage的并行度来提升性能，调度这样的小计算节点是非常浪费的，因为调度一台机器开启一个计算任务是需要调度时间和启动时间的，对于小任务，这样的时间并不可以忽略；而对于系统来说，这样的小数据带来的随机上的额外开销也是一种低性能。我们经验性地认为这样的小数据子任务在一个大数据计算任务中所占比例如果多于50%，那么这个任务很有可能是的459个是潜在的性能低下的任务。其中有部分的任务是相同的代码重复提交处理不RecurringJob124个进行了深入的人工分析，通过观察运行时并行度的统计图和Vertex吞吐量的统计图，以及回放整个计算的过来发现性能瓶颈所在的地方，然后通过人工定位找到对应的用户代码，根据任务执行过中的日志记录，综合分析整个执行过中用户层面和系统层面SCOPE平台上进行了重进行。因为大部分任务的运行时间很长，而且我们的研究工作不能影响产品部门正常在任务完成后就会被系统自动删除，于是无法获得这部分的数据进行实验，因此这部分的任务没有办法被重新运行。本文研究中所的大数据任务均是微软工师编写的，来自微软多个产品部门。一个在真实应用场景下的产品组集群中广泛采样的数据集。我们发现，因为商业团队工普遍包括多个MapReduce计算过这可能是由于微软超大规模的复杂系统应用的需MapReduce实现的大数据集群中是较为少见象用SOPE和#编写运行在微软大数据平台上的序。然而，我们发现的三个层面的导致任务性能低效的原因在其他的大数据平台上也是存在pdue现Hdoop次Hive和PigSLONT，SMDISTINT样的重语，并支持过语言定的F我发采了独特特性的编方影响任务的并行性能。本文的研究目标是在对于任务执行过深入了解的基础上，找出导致用户任务性能不佳的原因，这些原因可能有很多种，但是我们不考虑客观条件的限制因素导致的任务性能低下，比如说用户的权限较低导致其只能在系统中申请到比较少的额定并行资源，进而用户的任务受限于这个额定并行资源的量并在最终表现出性能不佳。这样的问题并不能通过代码优化和系统优化来解决，所以在这篇中我们不讨论类似这样的因素导致的性能问题。另外，因为同一台机器上运行的其他子任务对于我们研究的子任务造成了不良的影响，进而最终导致整个任务运行缓慢的场景在现实中是存在经有很多其他的研究工作在解决上面描述到的问题，这种研究对于分布式并行计算系统整体的性能的稳定性提升有着重要的研究价值，但是这些因素对于任务本身来说是大数据平台上的编实践和系统对于任务的优化工作。人工判断原因的人工的检查分析任务的性能是必不可少的，在整个研究过中，包括找到任务瓶颈SCOPE平台是一个复杂的大数据平台，平台的复杂性和任务的多样的任务，我们采用保守的方法，将其归类到不明原因的那一类，本文中一种主要的发现来阐述。对于每个任务，我们通过反复的确认来尽可能减少性带来的错误。我们通过上面的方法从大量的大数据任务中分类出一部分的潜在的低性能任务集合。因为产品组的任务中有很大一部分是urringob以处理新的数据，所以我们认为这样的任务是相似的，不需要重复的分析。我们分析124言、数据和系统三个层面的大数据平台上特有的原因。我们对于这些方面都进行了深入而细致的分析，总结了当前大数据平台研究和性能优化所的和。在语言层面上，大数据平台设计的初衷就是为了给大量的用户一个简单的方法处理海量规模的数据，所以大数据平台采用隐藏分布式并行细节的编语言，大数据平的SQlike语言既符合数据处理的语义又容易编而过型语言编写的用户定（F的差异。我们发现用户一些不良的写代码的技巧会导致他们经常写出一些对于底层优化器不友好的代码，无法应用一些编译优化，从而导致整个任务的性能显著降低。SPESOPE的语义让用户能够很简单清晰的完成他们的业务逻辑的编写，但其中有些用户代码对于底层系统的优化器来说是不友好的，它潜在的破坏了优化器的一些假定，从而导致一些并行优化没有办法应用，这最终导致了整个任务呈现出性能不佳的现象。而从另代码是有问题，甚至他们都无法分辨出来这样的过中存在着性能不佳，唯一知道的就是任务运行的时间很长，这无疑是非常致命的。PartialAggregate这样的减少们的优化后运行时间减少了64%，获得了极大的性能提升。在SCOPE图 种例子中，用户代码中的COUNTDISTINCT操作会导致同时进行的其他聚合操作的PartialAggregate优化失效，使得代码一阶段完成后没法通过PartialAggregate进行部分的聚合来减少数据项，从而使得DataSkew的现象更加的严重。PartialAggregateMapReduce计算模型中的重要优化，如第2章中所示，它是指对于在Reducer阶段进行的满换和结合律的操作可以在Mapper阶段完成后shuffle阶段开始之Mapper对应的机器上先进行局aggregate，这样能显著减shuffle阶段的数据量。在SCOPE中绝大部分内建的聚合函数都是可交换和结合的。DtaSkw也如第2章中所示的那样，是指同一阶段中不同的计算节点间处理的数据量上存在差别的现象，而PrtilAgrgte优化不只是能显著减小hufle过的数DtaSkw的现象对于计算的影响，当上个阶段的输出存在很严重的SkwPrtialAgrgteduey一个了，那么下一阶段的输入数据项最多就是上一阶段的并行机器数那么多，这样能DtaSkw的影响。JobRecurringJob中存在上面描述的现象。RecurringJob是一种成产品任务，其代码通常是稳定的、保持不变的，但是会针对这里我们用这个RecurringJob某一次的运行结果来作为例子进行分析。图4.1是那一次的运行性能表现图，横坐标表示Job图 JobTaskJob运行过Vertex。根据图4.2JobVertex所属的SV11_AggregateStage163.91GB51.54GB的数据，其并750Vertexdataskew的问题，也就是分配时72%VertexVertex执行的时间过长，直接导致了整个Job的执行运行时间被大大延迟。虽然这个任务的低性能表现出来是一个DataSkew的现象，但是我们发现其中的编译优化后的代码，将执行过的Stage跟用户代码进行对应，从而找到导致该现象产最终，我们发现导致全局性能缓慢的Stage是一个REDUCE过为了更好地抽JobSUM操SELECTaasSELECTaasCOUNT(DISTINCTb)asB,SUM(c)asC,SUM(d)as…FROM123456SCOPE是这样处理这段代码的，一个PROCESS（相当于MapReducer中的Mapper）的阶段，每一个PROCESSORtable这个表格按照AGroupBy操DISTINCT操作PROCESSOR没有办法对于CDPartialAggregate的操作。然后REDUCE阶段的每个节点通过shuffle操作，从每一个Mapper中获取自己ReduceKeyA这个column上相同的数据项放到一起，并按照B排序，之后就可以进行对应的COUNTSUM（c在这个例子中，很不幸的是，输入的数据存在严重的数据倾斜（dataskew）问题，APartialAggregate导致数据没有显著的减少，大量的数据在shuffle过中将被放置到同一个Vertex上面，进而导致处理这个值的Vertex重负，这也就是导致了它运行的时间占掉了整个Job执行时间的72%的原因，从而这个Vertex变成整个Job的瓶颈。PartialAggregate的操作PartialAggregatede操作拆分开，分别处理，最后再合并起来。于是可以把序改写成下面的形式，从而避免这个问题，提升并行的效率，减少Job的执行时间。这样，对于T1的操作来说，SUM操作是满换律和结合律的，系统会通过PartialAggregate优化shuffle过最后Mapper上的输出对于特定的A值只会有一行的数据，这样数据量大大减小，即便有DataSkew也能显著的减少其影响。对于T2的操作，虽然没有PartialAggregate但是shuffle过不需要对于C和D这T2的操作时可以并行执行的，这样也对于执行效率有显著的提升。T1T1=SELECTaasSUM(c)asC,SUM(d)as…FROMT2=SELECTaasCOUNT(DISTINCTB)asFROMT3=SELECTT1.*,T2.BFROMT1,T223456789我们用修改后的方法重新运行了这个序，运行时间减少了64%，获得了很好的在这个例子中，SELECTcolumnDISTINCT操作，导致了SELECTcolumnPartialAggregate优化，这增大shuffle过的网络IO，更不幸的是这个例子还遇到DataSkew的现象，没有ParticalAggregate最终导致这个现象更加的严重。然而系统也不能简单的自动进行类似上面形式的优化，因为如果Acolumn数据量很大的话，改写成后面这种形式反而会增大我们认为，DISTINCT操作在达到语义上便捷的同时对于分布式的运行环境来说是有不良影响的，因为同其他的聚合函数的结合使用会导致这个操作不满换律和结合columnDataSkew带来不良的影响，让整个Job表现出相当差的性能。不只是DISTINCT和其他聚合函数的组合，用户定义的不满换律或者结合律的聚合函数也会带来一样的问题。进一步的，我们认为应该重新慎重的考虑大数据平台上编语言的设计，编语言不仅要能够让用户高效灵活的表达自己的处理逻辑，而且对于系统优化器也应该是友好的，这样才能做到编灵活性和任务执行效率上的兼顾。从另一方面，在当前的编语们对于这种分布式数据并行系统实现和优化的了解，我们可以找到的不友好的模式，大数据平台上采用数据分块的方式来使得同一阶段中的子任务可以并行的执行，广泛性和任务的复杂性导致了任务过中存在多个运行的阶段，对于中间阶段我们很难控制其均匀性。尽管我们对于数据不平衡的操作都提供了一些更好的方法来避免，但是这需要用户的信息输入，并且在代码中给出适当的提示，这样才能帮助系统优化器优化整个执行过减少整个任务其他正常运行部分的间接开销，保持高效率。我们的研究中发现，除了最常见的GroupBy后数据不均匀导致的Skew现象，还有JOIN操作带来的Skew现象，这是一种更加棘手的Skew现象，因为是输出的数据存在的现象，甚至SCOPE开发都没有这个现象的存在，连事后分析的Skew现象的系统都没有关于JoinSkew现象的反馈机制。Join是数据库中非常普遍的操作，SCOPE也支持这样的操作。Join操作将有两个输入数据集，而且Join操作是接的操作，有可能会导致数据量的增大，甚至有的时候会导致数据变得异常的大，最的情况莫过于Join的操作是一个笛卡儿积（CrossJoin）若两个输入表的行数nm，那么输n*m行。SkewJoindataskew的一种Join中Join操作连Key存在Vertex这个现象最麻烦的地方在于系统难以预计JoinVertex也有可能有极大的输出数据，这样导致了调度上的和任务性能上的问题在SCOPE系统中，一般默认的把两个非常大的表连接起来的方法是先让这两个输这里面每个数据项会通过这个分块函数计算出一个值，称为PartitionKey，分块函数保证JoinKey相同的数据他们的PartitionKey也是相同的，之后系统通过shuffle过将具有相同的Key值的数据项放到相同的Vertex上面处理。连接操作的JoinKey在同一个Vertex上可能分配不一样。比如说对于不同的两个VertexVertex处理partitionkey是aVapartitionkeybVbnm个数VertexJoinKeyVa1JoinKey是x，Va2JoinKey是yJoin操作后没有任何输出，而对于Vb来说，Vb1Vb2JoinKeyzn*m行。显然，数据量大的节点会花费的时间，导致整个任务在相当长的一段时间内处于现有SCOPE的机制是不允Vertex运行超5个小时，JoinSkew现象会导Vertex有更大的可能会失败，也会导致失败恢操作将产生比输入还要多很多的输出，这些输出的结果将写在该rtex磁盘中，长时间大数据量的写入将对机器的磁盘带宽产生影响，这很可能会影响在同rtex的执行。个Job运行失败了。任务失败原因是，有几个Vertices运行超过5个小时导致整个JobJoinColumnStage使用。Join操作的两个输入分别SV2，SV3的输出，大小153GB215MB，数据partition1541G左右的数据。SV45个小时后被系统停止图 SkewJoinJob部分执行表 ExitProcessexecutiontimeistooProcessexecutiontimeistooProcessexecutiontimeistooProcessexecutiontimeistooVertices5SCOPE系统是不Vertex执行超5个小时的，于是整Job被系统终止。表1SV4Stage的Vertex按照运行时间降序排序的前5个Vertex的详细情况，VertexName一列括JMVertex运行缓慢进而克隆新Vertex来避免过长的运行时间（这个优化在这里显然也没有起到效果。我们发现运行超过5个小时的Vertices输入数据都不大，均小于0.5G，比平均值1G还小，但是都产生了上百G的输出，远高于所有Vertex平均的13.43G输出。joinSet=joinSet=SELECTFROMtable1ONtable1.x==12345严重的倾斜，导致Vertex运行时间过长，进而被SCOPE系统终止。可以让用户指定要求系统检测Join过中可能的Skew问题。[ [ 1oinSkwoin的方Hit系统去检测数据倾斜的现象是否存在，并采用更合适的执行计划来对待这部分倾斜数据，当然这会引入一些额外的开销，但是在这个例子中完全足以抵消Skw带来的不良象如用想要好控这个Skwoin照SkwoinHntprtitionprtition的大小和个数。SOPEoptmir会根据oin用户向系统提示这个Join操作存在严重的数据倾斜的情况，系统会根据用户提示DAG图，用更好的SkewJoinHint后，整个任务得以顺利的执行，从之前的运行将近7个小时后失败变成了优化后3.9个小时的时间完成。数据：数据存在严重的Skew，就是说某个特定PartitionKey上的两个输入有相当JoinKeyJoin操作会在数据量上有成倍的增长，因此产生非常多的输出，导致输出的IO成为了瓶颈。据具有怎样的分布情况，是不可知的，一方面用户没有告知系统相关的信息，另一方面由系统自发的去检测数据模式则会带来严重的性能问题，在大多数情况图 优化后部分的Job执行下是得不偿失的。Join操作隐含着带来的数据的特性，该操作的输出数据大0.5GB的输入居然能产生375GB的输出，从而让这个Vertex成为整个序的瓶颈。正是因为系统无法从输入的数据大小上判断出Join操作是否存在Skew，所以采用了默认的Join的Join方法只不过系统没法采用。数据某些度上的不均衡，会导致中间过存在急剧膨胀的现象，有可能使得一个Vertex需要处理过多的数据，从而对于序性能的危害。oin带来任务性能不佳的问题，有的时候大数据量的本地磁盘写入操作还会影响在这个机器上运行的其他操作，从而复杂的性能问题。对于序员来说，要有数据跟并行计算任务性能密切相关的认识。在编写序的时候要慎重地使用Join操作，同时在提交任务的时候要确认自己处理的数据的数据模式，相对的，在Join操作中对于分布不是很均匀的时候，要根据输入数据依据数据量的大小和倾斜度，配合系统的Hint机制给出的信息，整个执行过的优化，这供的信息处理好数据层面的问题，至少是一些有效的反馈机制。在产品组反复运行的任务这种背景下，反馈机制可以帮助用户改进RecurringJob的性能。大数据平台是一个现实中不断迭代发展中的软件系统，系统不可避免的存在一些任务有可能违背了系统在设计之初的某些假定，从而导致了一些不良性能的现象。对于一个并行序来说，并行度是很重要的决定并行序性能的指标。然而遗憾的MapReduce模型中，REDUCEMAP阶段开始前就决定，因MAPREDUCE阶段的任务数进行重新分块，但是这个数值很难很好的确定，因为一个并行阶段的效率跟它的输入数据量和并行任务PEDE阶段的输入数据量的。SOPEDrydpdue的优化。它有一个动态调整并行任务个数的机制，可以在执行的过中根据一些信息对于阶段任务数进行动态调整，但是其本质也是不知道上一个阶段的实际输出的，只能通过预估的方式进行调整，不可避免的有其限制性。SOPE出于效率的考虑，对于EDE和OINE25样的大数据任务都会在开始伴随着一个大量的筛选过滤的过实际任务处理的数据量通常比输入数据要小很多，如果这个上限设置的比较高，会导致有很多rtis只处理极小量的数据，系统底层被迫创建很多的小文件，这样会有很大的上的间接开销，从而性能表现不好。并且如果上一个阶段有m个并行任务，这一个阶段设置了n的hufle过会有mn个连接，如果m和n都很大话，为一个并行任务的瓶颈，特别是针对那种中间过也是海量数据的计算复杂型任务的时ReducerReducer阶段持续很长的时间。行计算资源并不是的，这些限决于调度机制、用户权限以及其他的一些因素。Allocation了任务运行时必须部分的并行资源，另一方面也限制任务并行度不能无，同时它也是系统优化器在确定任务执行流图的时候会考虑的一个因素。当可以并行执行的任务数味着单个任务处理的数据量过小，少到几Mk这样级别的数据，这导致了系统不得不创建很多的小文件，子任务数据的时候几乎都是碎片般的随机，这将在存作为介质的，大量的随机小文件使得时间都花在机械硬盘的寻道上，从而延长并行度过高这种现象一般是产生在PROCESSREDUCECOMBINE这两个阶段，SCOPE图 Job运行性能表现限制，而对于PROCESS阶段没有任何的限制，开始阶段的PROCESSOR个数完全取决于数据量和datapartitionPROCESSOR并行度主要取决于上一个阶段的Vertex数。研究过中发现了这样一种情况，当前一个阶段处理完一个数据集后，该数据集POESS阶段来执行的。具体的,rtexpror均处于可以被调度的状态，接下来调度器基于当前可以调度的资源量并且综合考虑数据的所在地（olityrtisrtexrties完成之后，调度系统才知道这个阶段的具体输出数据量大小；而实际上在系统实现上，通常为了利用好空rtexrtie，这rtis度器发现初始阶段产生的数据量很小，不应该使用这么多的并行任务，但是也已经来不及了。于是我们在最后会看到系统安排了大量的rtis处理极少量的数据，这种并行度过高的现象显然也是一种性能不佳的体现。图4.5是我们中发现的一个Job的运行性能表现图，横坐标是时间，单位是秒，纵坐标是并行运行的Vertex个数，不同颜色的色块代表不同的Stage，具体如图例图 局部的Job运行执行这里是一个Combiner数量不合适的例子，当然Aggregator（也就是Reducer）同理也存从图4.5StageSV7_Combine_Split运行了相当长的时间，显然是整个任务的瓶颈。SV7Job263s左右就开始，但是在2500s的时候才结束，占总运行时间的85%以上，运行时间非常的长。图4.6JobSV7的输入数据SV3SV6的输出，两REDUCE阶段。SV7在输378.43GB数据后Join1.14TBSV7将数据按照列进行切片，方便下面的Stage使用。图4.7SV7Stage的详细执行图，我们可以看出，SV7首先是对于输入的数据进行了Join的操作，然后通过个叫MobileIEDecoder的用户定义函数，对于数据进行图4.8SV7Vertex读写数据量的统计图，横坐Vertex的个数，总共有250个，纵坐标表示数据量的大小，单位是GB，蓝色的是读数据的量，红色的是写数据的量。我们从中可以看出，SV7这个Stage处理的数据量是比较均匀的，我们对于这个UDF人工的进行静态代码分析，发现这是一个复杂的字符串图 SV7的详细执行图 一个复杂度比较高的用户定义函数，而这导致了整个Stage运行缓慢，整个Job的瓶颈只被安排了250个Vertices，也就是说它的最高运行时并行度只能有250。虽然这个Job的TokenAllocation高达1068，但是多于250那部分的计算资源没有用到。这显然也是Job的TokenAllocation1068。也就是说这Job可以获1068的受系统保证的并行度。但是因为SV7250的并行度，Job长期处于并行250的状态，大大浪费了申请到的资源。并且从执行图可以看出，SV7Vertex都至少花了1000s才完成。完全可SV7Vertex处理的数据量进而减少整个Job的运行时间。这是要求用户对于自己处理的数据量要有一个比较准确的估计，而对于序中的Hint是有可能会没有作用的，因为优化器分析后可能会发现用户要求的计划是做不到的，那么这个PLANHINT就无法生效。有一个产品组任务，其输入大约是264TB数据，首先是一个数据转换和过滤的初始阶段，SCOPE40000Vertices进行这部分的处理，这是合理的。然而如图4.9380MBPROCESS过将对于这个数据进行处理，分别是SV17SV28，但是每个阶段都安排了40000个图 Job执行流Vertices，而这个时候我们知道并行的意义已经不大了，这种情况下高并行对于和调度系统来说的间接开销之大已经显而易见了。同实例一一样，PLANHINT可以帮助解决这个问题，用户通过给系统提示，限制PROCESS阶段的任务数，也减小了后继阶段的上的间接开销，这在提升任务性能但是这样的解决办法还是无法解决最开始阶段过滤后产生小量数据给系统带来的这部分间接开销。除非系统中存在对于极小量的输出将其放在内存中，让后面阶的rtex直接读内存，这样就能完全避免磁盘随机所损失的时间。并将得到的数据直接输出到分布式的文件系统中。第二个任务再读这个输出的中间数据，利用了分布式文件系统的性质把数据组织成比较大的文件块。这样后面的第二个任务就减少了很多的调度和上的间接开销，其需要调度的任务数也少了，数据的过也变成对于大文件的顺序了，从而获得良好的性能表现。特别是对于那种过滤后的中间数据将被反复使用的情况，第二种解决方更加的有效。PROCESS、REDUCE和COMBINE的阶段都支持采用（UDF在MapReduce系统中，Reducer的个数必须在Mapper开始前就决定，因为MapperReducerMapper完成前，系统又的数据大小来进行估计的，这带来了系统决策上的。另外，在SCOPE这样的基于DAGMapReduceREDUCEVertex个数的设置是具有全局影响的，这显然比单独的、输入数据量是已知的一个MapReduce的情况更加SCOPE系统能够做到的动态调整又非常的有限，当然这种动态调整本SOPE对于EDE20务都会在开始伴随着一个大量的筛选过滤的过实际任务处理的中间过数据量通被迫创建很多的小文件，这样会有很大的上的间接开销。但是随着现在大数据平台更加广泛的使用，这样的限制现在已经不适合很多的应用了，现在甚至有任务是输入的时候只比较少的数据，但是中间的数据会很大，这样显然违背了平台设计的时候潜在的假定。于是，在当前复杂的应用任务的需求下，采用动态并行度决策的机制是很有必要的，但是也是很有难度和的。实时的获取运行时的任务信息并进行一些调整肯定会带来系统的额外开销，如何在尽可能小的开销下完成更好的动态决策是一个非常有性的问题。在产品组集群中，我们可以结合产品组任务反复运行的特点，根据任务Kvulyatl.[9]和itl.[10]的工作，均对类似的分布式系任务进行了分析和总结。前者是分10Hdoop集群的log.%Hdoop任务消耗了系统的计算资源，因此减少这样可以有效的增加系统资源，但是前者分析的数据源是来自大学的大数据研究团队的数据，数据量较小而且跟现实的产品SOPE产品组的日常任务进行了分析，分析中250失任，了些类失原解方。们SOPE系统因为这些错误的代码而损失的计算性能。但是他们的工作只分析了失败的任务，而没有分析运行缓慢性能不佳的任务。行计算领域非常的问题，很多工作对于这个问题进行了深入的研究，尝试从系统优化的角度，对于这任务进行更好的分析。这些工作包括Mantri[11]，SkewTune[12]，Scarlett[13]，还有应用了任务克隆的[14]、任务推测的[15]。MantriScarlett都是通过对于skew的原因进行建模分析来优化调度的过从而提升性能。Mantri使得在调度的时候考虑当前资源现状，从而减少出现Skew的概率；Scarlett则通过分析和预测活跃数据所况。SkewTuneSkewDataSkew的，但是消耗了的系统资源。这些工作都有效的优化了系统，来减少Skew现象的出现和降低Skew现象的影响。SCOPE从代码编译到优化执行到调度过都有一系列的系统优化的工作，使得整个系统以及运行的大数据任务更加的高效。除了SCOPE[16]提到的优化，近几年还有很多在这个平台上的优化工作。Ananthanarayananetal.[11]的工作通过探测发现长时间没有完成的子任务，也就是Outlier，通过重启任务、根据网络环境调度任务和保护有价值的任务输出来减少Outlier对于序的影响，加快并行序的运行。Zhangetal.[17]SUDO的优化框架，可以对用户定义的函数进行分析，使其对于优化器来说不再是黑盒，从而采取对应的优化，减少data-shuffle过中的网络IO。Guoetal.[18]的工作通过分析序代码，进行相应的代码优化，采取删掉没有必要的代码和数据、更早的进行数据过滤等操作，减少并行序执行过中最耗费网络IO的data-shuffle过的数据量，从而提高SCOPE序的执行效率。这些优化在编译阶段和运行阶段都对于SCOPE序的执行进行了深度的优化，从各个方面提高了分布式执行计划的效率。但是即便有这么多的优化，我们还是发现了运行的序中存在这些性能动态调度方面，Optimus[Dryad[8CIEL[20]上的执行图动态重写的机制，结合了收集到的数据信息统计以及次型语言的语义，获得了远高于静态产生的执行计划的性能表现。但是我们注意到Optimus并没有做到对于用户定义函数语义的分型SQlike语言和过型语言#va层pdue系统的应用构成的大数据平台目前已经得到了广泛的应用，给用户在海量机器平台上实现并行计算序带来了巨大的便利，使其可以不用考虑繁杂的并行计算的细节，切实的提高了编的效率。对于这样的大数据平台进行性能研究工作有着重要而深远的意义。分析工作，抓住了海量数据环境下大规模并行计算序的特征，找到有效其中一部分性能表现不佳的SCOPE任务经过修改和重新运行后获得了性能上64%的任务运行时间，并且大大减少了任务因为运行时这只是一个刚刚开始的工作，了解大数据平台上目前任务的性能瓶颈的根源所在可以帮我们更好的了解大数据系统当前的根本问题。本文的研究工作也让我们发现了在语言、数据和系统层面上还有很多的和等待我们去突破和解决，这都是一些可以进行的进一步工作。包括：语言层面上，设计大数据编语言的时候，在保持其灵活性和透明性的同时应数据层面，因为数据平衡是一件重要而的问题，系统设计时应引入相应的制，鼓励和引导用户处理好数据层面的问题，至少是一些有效的反馈机制。在产品组任务的背景环境下，可以通过有效反馈来帮助用户改善反复运行的Recurringob的性能。分析来提供一种动态的执行过优化机制，动态调整不同阶段的任务运行的并(SKSDE)(SA)，让我够在大三暑期和大四期间学习到分布式系统领域的相关知识，并且在一个真实的大数据平台上完成。感谢导师李未教授，在百忙之中仍抽出时间对于毕业设计进行指导，在在答辩过中的帮助和指导。感谢罗杰老师指导我的写作。感谢我在微软亚洲的ntr，首席研究员周礼栋，在科研思维上面对于我学态度和对于科研前沿的敏锐洞察力让我受益匪浅。感谢一起参与到这个项目中的杨。感谢辅导员闫昭和，大学四年来从学习到生活上的关心和帮助，让我远离家乡也倍感。在SA实习期间认识的同组聊天触发了我很多的思考，让我对计算机科研领域有了更加广泛而深入的了解，感谢他们在科研和生活上给予帮助和鼓励。感谢女朋友陈倩岚对于研究工作一直以来的理解支持和无微不至的照顾。最后感谢父母23年来的教育和，这四年来在远方对支持和鼓励，没有就没有今天。，这篇文章乃至这整个毕业设计的完成就是对付出DeanJ.,GhemawatS.MapReduce:SimplifiedDataProcessingonLargeClusters[J].Commun.ACM,2008,51(1):107–113..OlstonC.,ReedB.,SrivastavaU.,etal.Piglatin:anot-so-foreignlanguagefordatapro-cessing[A].Proceedingsofthe2008ACMSIGMODinternationalconferenceonMan-agementofd]..[S.l.]:[s.n.],2008:1099–1110.GatesA.F.,NatkovichO.,ChopraS.,etal.Buildingahigh-leveldataflowsystemontopofMap-Reduce:thePigexperience[J].ProceedingsoftheVLDBEndowment,2009,ThusooA.,SarmaJ.S.,JainN.,etal.Hive:awarehousingsolutionoveramap-reduceframework[J].ProceedingsoftheVLDBEndowment,2009,2(2):1626–1629.ThusooA.,SarmaJ.S.,JainN.,etal.Hive-apetabytescaledatawarehouseusinghadoop[A].DataEngineering(ICDE),2010IEEE26thInternationalConferenceon[C]..[S.l.]:[s.n.],ChambersC.,RaniwalaA.,PerryF.,etal.FlumeJava:easy,efficientdata-parallelpipelines[A].ACMSigplanNotices[C]..[S.l.]:[s.n.],2010，45:363–375.ofMassiveDataSets[J].Proc.VLDBEndow.,2008,1(2):12