Spark性能优化——和shuffle搏斗

标签: Distributed System shuffle Spark | 发表时间:2016-05-21 18:48 | 作者:四火
出处:http://www.raychase.net

Spark性能优化——和shuffle搏斗

Spark的性能分析和调优很有意思,今天再写一篇。主要话题是shuffle,当然也牵涉一些其他代码上的小把戏。

以前写过一篇文章,比较了 几种不同场景的性能优化,包括portal的性能优化,web service的性能优化,还有Spark job的性能优化。Spark的性能优化有一些特殊的地方,比如实时性一般不在考虑范围之内,通常我们用Spark来处理的数据,都是要求异步得到结果的数据;再比如数据量一般都很大,要不然也没有必要在集群上操纵这么一个大家伙,等等。事实上,我们都知道没有银弹,但是每一种性能优化场景都有一些特定的“大boss”,通常抓住和解决大boss以后,能解决其中一大部分问题。比如对于portal来说,是页面静态化,对于web service来说,是高并发(当然,这两种可以说并不确切,这只是针对我参与的项目总结的经验而已),而对于Spark来说,这个大boss就是shuffle。

首先要明确什么是shuffle。Shuffle指的是从map阶段到reduce阶段转换的时候,即map的output向着reduce的input映射的时候,并非节点一一对应的,即干map工作的slave A,它的输出可能要分散跑到reduce节点A、B、C、D …… X、Y、Z去,就好像shuffle的字面意思“洗牌”一样,这些map的输出数据要打散然后根据新的路由算法(比如对key进行某种hash算法),发送到不同的reduce节点上去。(下面这幅图来自 《Spark Architecture: Shuffle》

Spark性能优化——和shuffle搏斗

为什么说shuffle是Spark job的大boss,就是因为Spark本身的计算通常都是在内存中完成的,比如这样一个map结构的RDD:(String, Seq),key是字符串,value是一个Seq,如果只是对value进行一一映射的map操作,比如(1)先计算Seq的长度,(2)再把这个长度作为元素添加到Seq里面去。这两步计算,都可以在local完成,而事实上也是在内存中操作完成的,换言之,不需要跑到别的node上去拿数据,因此执行的速度是非常快的。但是,如果对于一个大的rdd,shuffle发生的时候,就会因为网络传输、数据序列化/反序列化产生大量的磁盘IO和CPU开销。这个性能上的损失是非常巨大的。

要减少shuffle的开销,主要有两个思路:

  1. 减少shuffle次数,尽量不改变key,把数据处理在local完成;
  2. 减少shuffle的数据规模。

先去重,再合并

比如有A、B这样两个规模比较大的RDD,如果各自内部有大量重复,那么二者一合并,再去重:

A.union(B).distinct()

这样的操作固然正确,但是如果可以先各自去重,再合并,再去重,可以大幅度减小shuffle的开销(注意Spark的默认union和Oracle里面的“union all”很像——不去重):

A.distinct().union(B.distinct()).distinct()

看起来变复杂了对不对,但是当时我解决这个问题的时候,用第二种方法时间开销从3个小时减到20分钟。

如果中间结果rdd如果被调用多次,可以显式调用cache()和persist(),以告知Spark,保留当前rdd。当然,即便不这么做,Spark依然存放不久前计算过的结果(以下来自 官方指南):

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.

数据量大,并不一定慢。通常情况下,由于Spark的job是放到内存里面进行运算的,因此一个复杂的map操作不一定执行起来很慢。但是如果牵涉到shuffle,这里面有网络传输和序列化的问题,就有可能非常慢。

类似地,还有filter等等操作,目的也是要先对大的RDD进行“瘦身”操作,然后在做其他操作。

mapValues比map好

明确key不会变的map,就用mapValues来替代,因为这样可以保证Spark不会shuffle你的数据:

A.map{case (A, ((B, C), (D, E))) => (A, (B, C, E))}

改成:

A.map{case ((B, C), (D, E)) => (B, C, E)}

用broadcast + filter来代替join

这种优化是一种特定场景的神器,就是拿大的RDD A去join一个小的RDD B,比如有这样两个RDD:

  • A的结构为(name, age, sex),表示全国人民的RDD,超大
  • B的结果为(age, title),表示“年龄 -> 称号”的映射,比如60岁有称号“花甲之年”,70岁则是“古稀之年”,这个RDD显然很小,因为人的年龄范围在0~200岁之间,而且有的“年龄”还没有“称号”

现在我要从全国人民中找出这些有称号的人来。如果直接写成:

A.map{case (name, age, sex) => (age, (name, sex))}
 .join(B)
 .map{case (age, ((name, sex), title)) => (name, age, sex)}

你就可以想象,执行的时候超大的A被打散和分发到各个节点去。而且更要命的是,为了恢复一开始的(name, age, sex)的结构,又做了一次map,而这次map一样导致shuffle。两次shuffle,太疯狂了。但是如果这样写:

val b = sc.broadcast(B.collectAsMap)
A.filter{case (name, age, sex) => b.values.contains(age)}

一次shuffle都没有,A老老实实待着不动,等着全量的B被分发过来。

另外,在Spark SQL里面直接有BroadcastHashJoin,也是把小的rdd广播出去。

不均匀的shuffle

在工作中遇到这样一个问题,需要转换成这样一个非常巨大的RDD A,结构是(countryId, product),key是国家id,value是商品的具体信息。当时在shuffle的时候,这个hash算法是根据key来选择节点的,但是事实上这个countryId的分布是极其不均匀的,大部分商品都在美国(countryId=1),于是我们通过Ganglia看到,其中一台slave的CPU特别高,计算全部聚集到那一台去了。

找到原因以后,问题解决就容易了,要么避免这个shuffle,要么改进一下key,让它的shuffle能够均匀分布(比如可以拿countryId+商品名称的tuple作key,甚至生成一个随机串)。

明确哪些操作必须在master完成

如果想打印一些东西到stdout里去:

A.foreach(println)

想把RDD的内容逐条打印出来,但是结果却没有出现在stdout里面,因为这一步操作被放到slave上面去执行了。其实只需要collect一下,这些内容就被加载到master的内存中打印了:

A.collect.foreach(println)

再比如,如果遇到RDD操作嵌套的情况,通常考虑优化掉,因为只有master才能去理解和执行RDD的操作,slave只能处理被分配的task而已。比如:

A.map{case (keyA, valueA) => doSomething(B.lookup(keyA).head, valueA)}

就可以用join来代替:

A.join(B).map{case (key, (valueA, valueB)) => doSomething(valueB, valueA)}

用reduceByKey代替groupByKey

这一条应该是比较经典的了。reduceByKey会在当前节点(local)中做reduce操作,也就是说,会在shuffle前,尽可能地减小数据量。而groupByKey则不是,它会不做任何处理而直接去shuffle。当然,有一些场景下,功能上二者并不能互相替换。因为reduceByKey要求参与运算的value,并且和输出的value类型要一样,但是groupByKey则没有这个要求。

有一些类似的xxxByKey操作,都比groupByKey好,比如foldByKey和aggregateByKey。

另外,还有一条类似的是用treeReduce来代替reduce,主要是用于单个reduce操作开销比较大,可以条件treeReduce的深度来控制每次reduce的规模。

文章未经特殊标明皆为本人原创,未经许可不得用于任何商业用途,转载请保持完整性并注明来源链接 《四火的唠叨》

分享到:

相关 [spark 性能优化 shuffle] 推荐:

Spark性能优化——和shuffle搏斗

- - 四火的唠叨
Spark的性能分析和调优很有意思,今天再写一篇. 主要话题是shuffle,当然也牵涉一些其他代码上的小把戏. 以前写过一篇文章,比较了 几种不同场景的性能优化,包括portal的性能优化,web service的性能优化,还有Spark job的性能优化. Spark的性能优化有一些特殊的地方,比如实时性一般不在考虑范围之内,通常我们用Spark来处理的数据,都是要求异步得到结果的数据;再比如数据量一般都很大,要不然也没有必要在集群上操纵这么一个大家伙,等等.

Spark Shuffle过程分析:Map阶段处理流程

- - 简单之美
默认配置情况下,Spark在Shuffle过程中会使用SortShuffleManager来管理Shuffle过程中需要的基本组件,以及对RDD各个Partition数据的计算. 我们可以在Driver和Executor对应的SparkEnv对象创建过程中看到对应的配置,如下代码所示:. 如果需要修改ShuffleManager实现,则只需要修改配置项spark.shuffle.manager即可,默认支持sort和 tungsten-sort,可以指定自己实现的ShuffleManager类.

Spark性能优化指南——基础篇

- - 美团点评技术团队
在大数据计算领域,Spark已经成为了越来越流行、越来越受欢迎的计算平台之一. Spark的功能涵盖了大数据领域的离线批处理、SQL类处理、流式/实时计算、机器学习、图计算等各种不同类型的计算操作,应用范围与前景非常广泛. 在美团•大众点评,已经有很多同学在各种项目中尝试使用Spark. 大多数同学(包括笔者在内),最初开始尝试使用Spark的原因很简单,主要就是为了让大数据计算作业的执行速度更快、性能更高.

Spark性能优化指南——高级篇

- - 美团点评技术团队
继 基础篇讲解了每个Spark开发人员都必须熟知的开发调优与资源调优之后,本文作为《Spark性能优化指南》的高级篇,将深入分析数据倾斜调优与shuffle调优,以解决更加棘手的性能问题. 有的时候,我们可能会遇到大数据计算中一个最棘手的问题——数据倾斜,此时Spark作业的性能会比期望差很多. 数据倾斜调优,就是使用各种技术方案解决不同类型的数据倾斜问题,以保证Spark作业的性能.

实用教程|Spark性能优化之道——解决Spark数据倾斜

- - IT瘾-geek
实用教程|Spark性能优化之道——解决Spark数据倾斜.     2017-03-16 11:31  浏览次数:108. 为何要处理数据倾斜(Data Skew). 对Spark/Hadoop这样的大数据系统来讲,数据量大并不可怕,可怕的是数据倾斜. 数据倾斜指的是,并行处理的数据集中,某一部分(如Spark或Kafka的一个Partition)的数据显著多于其它部分,从而使得该部分的处理速度成为整个数据集处理的瓶颈.

Spark性能优化之道——解决Spark数据倾斜(Data Skew)的N种姿势

- - IT瘾-bigdata
本文结合实例详细阐明了Spark数据倾斜的几种场景以及对应的解决方案,包括避免数据源倾斜,调整并行度,使用自定义Partitioner,使用Map侧Join代替Reduce侧Join,给倾斜Key加上随机前缀等. 为何要处理数据倾斜(Data Skew). 对Spark/Hadoop这样的大数据系统来讲,数据量大并不可怕,可怕的是数据倾斜.

HADOOP SHUFFLE(转载)

- - 数据库 - ITeye博客
Shuffle过程是MapReduce的核心,也被称为奇迹发生的地方. 要想理解MapReduce,Shuffle是必须要了解的. 我看过很多相关的资料,但每次看完都云里雾里的绕着,很难理清大致的逻辑,反而越搅越混. 前段时间在做MapReduce job性能调优的工作,需要深入代码研究MapReduce的运行机制,这才对Shuffle探了个究竟.

[转]MapReduce:详解Shuffle(copy,sort,merge)过程

- - 芒果先生Mango的专栏
Shuffle过程是MapReduce的核心,也被称为奇迹发生的地方. 要想理解MapReduce, Shuffle是必须要了解的. 我看过很多相关的资料,但每次看完都云里雾里的绕着,很难理清大致的逻辑,反而越搅越混. 前段时间在做MapReduce job 性能调优的工作,需要深入代码研究MapReduce的运行机制,这才对Shuffle探了个究竟.

Spark概览

- - 简单文本
Spark具有先进的DAG执行引擎,支持cyclic data flow和内存计算. 因此,它的运行速度,在内存中是Hadoop MapReduce的100倍,在磁盘中是10倍. 这样的性能指标,真的让人心动啊. Spark的API更为简单,提供了80个High Level的操作,可以很好地支持并行应用.

Spark与Mapreduce?

- - 崔永键的博客
我本人是类似Hive平台的系统工程师,我对MapReduce的熟悉程度是一般,它是我的底层框架. 我隔壁组在实验Spark,想将一部分计算迁移到Spark上. 年初的时候,看Spark的评价,几乎一致表示,Spark是小数据集上处理复杂迭代的交互系统,并不擅长大数据集,也没有稳定性. 但是最近的风评已经变化,尤其是14年10月他们完成了Peta sort的实验,这标志着Spark越来越接近替代Hadoop MapReduce了.