Spark性能优化——和shuffle搏斗

   <p>Spark的性能分析和调优很有意思，今天再写一篇。主要话题是shuffle，当然也牵涉一些其他代码上的小把戏。</p>    <p>以前写过一篇文章，比较了 <a href="http://www.open-open.com/lib/view/open1455611177995.html">几种不同场景的性能优化</a> ，包括portal的性能优化，web service的性能优化，还有Spark job的性能优化。Spark的性能优化有一些特殊的地方，比如实时性一般不在考虑范围之内，通常我们用Spark来处理的数据，都是要求异步得到结果的数据；再比如数据量一般都很大，要不然也没有必要在集群上操纵这么一个大家伙，等等。事实上，我们都知道没有银弹，但是每一种性能优化场景都有一些特定的“大boss”，通常抓住和解决大boss以后，能解决其中一大部分问题。比如对于portal来说，是页面静态化，对于web service来说，是高并发（当然，这两种可以说并不确切，这只是针对我参与的项目总结的经验而已），而对于Spark来说，这个大boss就是shuffle。</p>    <p>首先要明确什么是shuffle。Shuffle指的是从map阶段到reduce阶段转换的时候，即map的output向着reduce的input映射的时候，并非节点一一对应的，即干map工作的slave A，它的输出可能要分散跑到reduce节点A、B、C、D …… X、Y、Z去，就好像shuffle的字面意思“洗牌”一样，这些map的输出数据要打散然后根据新的路由算法（比如对key进行某种hash算法），发送到不同的reduce节点上去。（下面这幅图来自 <a href="/misc/goto?guid=4959673489834239531" rel="nofollow,noindex">《Spark Architecture: Shuffle》</a> ）</p>    <p><img src="https://simg.open-open.com/show/72f5312efb70f94a45ebc0b82b2eb621.png"></p>    <p>为什么说shuffle是Spark job的大boss，就是因为Spark本身的计算通常都是在内存中完成的，比如这样一个map结构的RDD：(String, Seq)，key是字符串，value是一个Seq，如果只是对value进行一一映射的map操作，比如（1）先计算Seq的长度，（2）再把这个长度作为元素添加到Seq里面去。这两步计算，都可以在local完成，而事实上也是在内存中操作完成的，换言之，不需要跑到别的node上去拿数据，因此执行的速度是非常快的。但是，如果对于一个大的rdd，shuffle发生的时候，就会因为网络传输、数据序列化/反序列化产生大量的磁盘IO和CPU开销。这个性能上的损失是非常巨大的。</p>    <p>要减少shuffle的开销，主要有两个思路：</p>    <ol>     <li>减少shuffle次数，尽量不改变key，把数据处理在local完成；</li>     <li>减少shuffle的数据规模。</li>    </ol>    <p>先去重，再合并</p>    <p>比如有A、B这样两个规模比较大的RDD，如果各自内部有大量重复，那么二者一合并，再去重：</p>    <pre>  A.union(B).distinct()</pre>    <p>这样的操作固然正确，但是如果可以先各自去重，再合并，再去重，可以大幅度减小shuffle的开销（注意Spark的默认union和Oracle里面的“union all”很像——不去重）：</p>    <pre>  A.distinct().union(B.distinct()).distinct()</pre>    <p>看起来变复杂了对不对，但是当时我解决这个问题的时候，用第二种方法时间开销从3个小时减到20分钟。</p>    <p>如果中间结果rdd如果被调用多次，可以显式调用cache()和persist()，以告知Spark，保留当前rdd。当然，即便不这么做，Spark依然存放不久前计算过的结果（以下来自 <a href="/misc/goto?guid=4959615872009373530" rel="nofollow,noindex">官方指南</a> ）：</p>    <p>Spark also automatically persists some intermediate data in shuffle operations (e.g. reduceByKey), even without users calling persist. This is done to avoid recomputing the entire input if a node fails during the shuffle. We still recommend users call persist on the resulting RDD if they plan to reuse it.</p>    <p>数据量大，并不一定慢。通常情况下，由于Spark的job是放到内存里面进行运算的，因此一个复杂的map操作不一定执行起来很慢。但是如果牵涉到shuffle，这里面有网络传输和序列化的问题，就有可能非常慢。</p>    <p>类似地，还有filter等等操作，目的也是要先对大的RDD进行“瘦身”操作，然后在做其他操作。</p>    <p>mapValues比map好</p>    <p>明确key不会变的map，就用mapValues来替代，因为这样可以保证Spark不会shuffle你的数据：</p>    <pre>  A.map{case (A, ((B, C), (D, E))) => (A, (B, C, E))}</pre>    <p>改成：</p>    <pre>  A.map{case ((B, C), (D, E)) => (B, C, E)}</pre>    <p>用broadcast + filter来代替join</p>    <p>这种优化是一种特定场景的神器，就是拿大的RDD A去join一个小的RDD B，比如有这样两个RDD：</p>    <ul>     <li>A的结构为(name, age, sex)，表示全国人民的RDD，超大</li>     <li>B的结果为(age, title)，表示“年龄 -> 称号”的映射，比如60岁有称号“花甲之年”，70岁则是“古稀之年”，这个RDD显然很小，因为人的年龄范围在0~200岁之间，而且有的“年龄”还没有“称号”</li>    </ul>    <p>现在我要从全国人民中找出这些有称号的人来。如果直接写成：</p>    <pre>  A.map{case (name, age, sex) => (age, (name, sex))}   .join(B)   .map{case (age, ((name, sex), title)) => (name, age, sex)}</pre>    <p>你就可以想象，执行的时候超大的A被打散和分发到各个节点去。而且更要命的是，为了恢复一开始的(name, age, sex)的结构，又做了一次map，而这次map一样导致shuffle。两次shuffle，太疯狂了。但是如果这样写：</p>    <pre>  val b = sc.broadcast(B.collectAsMap)  A.filter{case (name, age, sex) => b.values.contains(age)}</pre>    <p>一次shuffle都没有，A老老实实待着不动，等着全量的B被分发过来。</p>    <p>另外，在Spark SQL里面直接有BroadcastHashJoin，也是把小的rdd广播出去。</p>    <p>不均匀的shuffle</p>    <p>在工作中遇到这样一个问题，需要转换成这样一个非常巨大的RDD A，结构是(countryId, product)，key是国家id，value是商品的具体信息。当时在shuffle的时候，这个hash算法是根据key来选择节点的，但是事实上这个countryId的分布是极其不均匀的，大部分商品都在美国（countryId=1），于是我们通过Ganglia看到，其中一台slave的CPU特别高，计算全部聚集到那一台去了。</p>    <p>找到原因以后，问题解决就容易了，要么避免这个shuffle，要么改进一下key，让它的shuffle能够均匀分布（比如可以拿countryId+商品名称的tuple作key，甚至生成一个随机串）。</p>    <p>明确哪些操作必须在master完成</p>    <p>如果想打印一些东西到stdout里去：</p>    <pre>  A.foreach(println)</pre>    <p>想把RDD的内容逐条打印出来，但是结果却没有出现在stdout里面，因为这一步操作被放到slave上面去执行了。其实只需要collect一下，这些内容就被加载到master的内存中打印了：</p>    <pre>  A.collect.foreach(println)</pre>    <p>再比如，如果遇到RDD操作嵌套的情况，通常考虑优化掉，因为只有master才能去理解和执行RDD的操作，slave只能处理被分配的task而已。比如：</p>    <pre>  A.map{case (keyA, valueA) => doSomething(B.lookup(keyA).head, valueA)}</pre>    <p>就可以用join来代替：</p>    <pre>  A.join(B).map{case (key, (valueA, valueB)) => doSomething(valueB, valueA)}</pre>    <p>用reduceByKey代替groupByKey</p>    <p>这一条应该是比较经典的了。reduceByKey会在当前节点（local）中做reduce操作，也就是说，会在shuffle前，尽可能地减小数据量。而groupByKey则不是，它会不做任何处理而直接去shuffle。当然，有一些场景下，功能上二者并不能互相替换。因为reduceByKey要求参与运算的value，并且和输出的value类型要一样，但是groupByKey则没有这个要求。</p>    <p>有一些类似的xxxByKey操作，都比groupByKey好，比如foldByKey和aggregateByKey。</p>    <p>另外，还有一条类似的是用treeReduce来代替reduce，主要是用于单个reduce操作开销比较大，可以条件treeReduce的深度来控制每次reduce的规模。</p>    <p> </p>    <p>来自： <a href="/misc/goto?guid=4959673489948580476" rel="nofollow">http://www.raychase.net/3788</a></p>    <p> </p>