Spark面对OOM问题的解决方法及优化总结(转载)

Spark中的OOM问题不外乎以下两种情况

• map执行中内存溢出
• shuffle后内存溢出

• execution内存是执行内存&＃xff0c;文档中说join&＃xff0c;aggregate都在这部分内存中执行&＃xff0c;shuffle的数据也会先缓存在这个内存中&＃xff0c;满了再写入磁盘&＃xff0c;能够减少IO。其实map过程也是在这个内存中执行的。
• storage内存是存储broadcast&＃xff0c;cache&＃xff0c;persist数据的地方。
• other内存是程序执行时预留给自己的内存。

execution和storage是Spark Executor中内存的大户&＃xff0c;other占用内存相对少很多&＃xff0c;这里就不说了。在spark-1.6.0以前的版本&＃xff0c;execution和storage的内存分配是固定的&＃xff0c;使用的参数配置分别是spark.shuffle.memoryFraction&＃xff08;execution内存占Executor总内存大小&＃xff0c;default 0.2&＃xff09;和spark.storage.memoryFraction&＃xff08;storage内存占Executor内存大小&＃xff0c;default 0.6&＃xff09;&＃xff0c;因为是1.6.0以前这两块内存是互相隔离的&＃xff0c;这就导致了Executor的内存利用率不高&＃xff0c;而且需要根据Application的具体情况&＃xff0c;使用者自己来调节这两个参数才能优化Spark的内存使用。在spark-1.6.0以上的版本&＃xff0c;execution内存和storage内存可以相互借用&＃xff0c;提高了内存的Spark中内存的使用率&＃xff0c;同时也减少了OOM的情况。

在Spark-1.6.0后加入了堆外内存&＃xff0c;进一步优化了Spark的内存使用&＃xff0c;堆外内存使用JVM堆以外的内存&＃xff0c;不会被gc回收&＃xff0c;可以减少频繁的full gc&＃xff0c;所以在Spark程序中&＃xff0c;会长时间逗留再Spark程序中的大内存对象可以使用堆外内存存储。使用堆外内存有两种方式&＃xff0c;一种是在rdd调用persist的时候传入参数StorageLevel.OFF_HEAP&＃xff0c;这种使用方式需要配合Tachyon一起使用。另外一种是使用Spark自带的spark.memory.offHeap.enabled 配置为true进行使用&＃xff0c;但是这种方式在1.6.0的版本还不支持使用&＃xff0c;只是多了这个参数&＃xff0c;在以后的版本中会开放。

OOM的问题通常出现在execution这块内存中&＃xff0c;因为storage这块内存在存放数据满了之后&＃xff0c;会直接丢弃内存中旧的数据&＃xff0c;对性能有影响但是不会有OOM的问题。

这种溢出的原因是在单个map中产生了大量的对象导致的&＃xff0c;例如&＃xff1a;rdd.map(x&＃61;>for(i <- 1 to 10000) yield i.toString)&＃xff0c;这个操作在rdd中&＃xff0c;每个对象都产生了10000个对象&＃xff0c;这肯定很容易产生内存溢出的问题。针对这种问题&＃xff0c;在不增加内存的情况下&＃xff0c;可以通过减少每个Task的大小&＃xff0c;以便达到每个Task即使产生大量的对象Executor的内存也能够装得下。具体做法可以在会产生大量对象的map操作之前调用repartition方法&＃xff0c;分区成更小的块传入map。例如&＃xff1a;rdd.repartition(10000).map(x&＃61;>for(i <- 1 to 10000) yield i.toString)。

面对这种问题注意&＃xff0c;不能使用rdd.coalesce方法&＃xff0c;这个方法只能减少分区&＃xff0c;不能增加分区&＃xff0c;不会有shuffle的过程。

数据不平衡除了有可能导致内存溢出外&＃xff0c;也有可能导致性能的问题&＃xff0c;解决方法和上面说的类似&＃xff0c;就是调用repartition重新分区。这里就不再累赘了。

这是我最近才遇到的一个问题&＃xff0c;因为hdfs中不适合存小问题&＃xff0c;所以Spark计算后如果产生的文件太小&＃xff0c;我们会调用coalesce合并文件再存入hdfs中。但是这会导致一个问题&＃xff0c;例如在coalesce之前有100个文件&＃xff0c;这也意味着能够有100个Task&＃xff0c;现在调用coalesce(10)&＃xff0c;最后只产生10个文件&＃xff0c;因为coalesce并不是shuffle操作&＃xff0c;这意味着coalesce并不是按照我原本想的那样先执行100个Task&＃xff0c;再将Task的执行结果合并成10个&＃xff0c;而是从头到位只有10个Task在执行&＃xff0c;原本100个文件是分开执行的&＃xff0c;现在每个Task同时一次读取10个文件&＃xff0c;使用的内存是原来的10倍&＃xff0c;这导致了OOM。解决这个问题的方法是令程序按照我们想的先执行100个Task再将结果合并成10个文件&＃xff0c;这个问题同样可以通过repartition解决&＃xff0c;调用repartition(10)&＃xff0c;因为这就有一个shuffle的过程&＃xff0c;shuffle前后是两个Stage&＃xff0c;一个100个分区&＃xff0c;一个是10个分区&＃xff0c;就能按照我们的想法执行。

shuffle内存溢出的情况可以说都是shuffle后&＃xff0c;单个文件过大导致的。在Spark中&＃xff0c;join&＃xff0c;reduceByKey这一类型的过程&＃xff0c;都会有shuffle的过程&＃xff0c;在shuffle的使用&＃xff0c;需要传入一个partitioner&＃xff0c;大部分Spark中的shuffle操作&＃xff0c;默认的partitioner都是HashPatitioner&＃xff0c;默认值是父RDD中最大的分区数,这个参数通过spark.default.parallelism控制(在spark-sql中用spark.sql.shuffle.partitions) &＃xff0c; spark.default.parallelism参数只对HashPartitioner有效&＃xff0c;所以如果是别的Partitioner或者自己实现的Partitioner就不能使用spark.default.parallelism这个参数来控制shuffle的并发量了。如果是别的partitioner导致的shuffle内存溢出&＃xff0c;就需要从partitioner的代码增加partitions的数量。

在standalone的模式下如果配置了--total-executor-cores 和 --executor-memory 这两个参数&＃xff0c;但是没有配置--executor-cores这个参数的话&＃xff0c;就有可能导致&＃xff0c;每个Executor的memory是一样的&＃xff0c;但是cores的数量不同&＃xff0c;那么在cores数量多的Executor中&＃xff0c;由于能够同时执行多个Task&＃xff0c;就容易导致内存溢出的情况。这种情况的解决方法就是同时配置--executor-cores或者spark.executor.cores参数&＃xff0c;确保Executor资源分配均匀。

这个比较特殊&＃xff0c;这里说记录一下&＃xff0c;遇到过一种情况&＃xff0c;类似这样rdd.flatMap(x&＃61;>for(i <- 1 to 1000) yield ("key","value"))导致OOM&＃xff0c;但是在同样的情况下&＃xff0c;使用rdd.flatMap(x&＃61;>for(i <- 1 to 1000) yield "key"&＃43;"value")就不会有OOM的问题&＃xff0c;这是因为每次("key","value")都产生一个Tuple对象&＃xff0c;而"key"&＃43;"value"&＃xff0c;不管多少个&＃xff0c;都只有一个对象&＃xff0c;指向常量池。具体测试如下&＃xff1a;

这个例子说明("key","value")和("key","value")在内存中是存在不同位置的,也就是存了两份,但是"key"&＃43;"value"虽然出现了两次,但是只存了一份,在同一个地址,这用到了JVM常量池的知识.于是乎,如果RDD中有大量的重复数据,或者Array中需要存大量重复数据的时候我们都可以将重复数据转化为String,能够有效的减少内存使用.

这一部分主要记录一下到spark-1.6.1版本&＃xff0c;笔者觉得有优化性能作用的一些参数配置和一些代码优化技巧&＃xff0c;在参数优化部分&＃xff0c;如果笔者觉得默认值是最优的了&＃xff0c;这里就不再记录。

这里需要稍微讲一下RDD和DataFrame的区别。RDD强调的是不可变对象&＃xff0c;每个RDD都是不可变的&＃xff0c;当调用RDD的map类型操作的时候&＃xff0c;都是产生一个新的对象&＃xff0c;这就导致了一个问题&＃xff0c;如果对一个RDD调用大量的map类型操作的话&＃xff0c;每个map操作会产生一个到多个RDD对象&＃xff0c;这虽然不一定会导致内存溢出&＃xff0c;但是会产生大量的中间数据&＃xff0c;增加了gc操作。另外RDD在调用action操作的时候&＃xff0c;会出发Stage的划分&＃xff0c;但是在每个Stage内部可优化的部分是不会进行优化的&＃xff0c;例如rdd.map(_&＃43;1).map(_&＃43;1)&＃xff0c;这个操作在数值型RDD中是等价于rdd.map(_&＃43;2)的&＃xff0c;但是RDD内部不会对这个过程进行优化。DataFrame则不同&＃xff0c;DataFrame由于有类型信息所以是可变的&＃xff0c;并且在可以使用sql的程序中&＃xff0c;都有除了解释器外&＃xff0c;都会有一个sql优化器&＃xff0c;DataFrame也不例外&＃xff0c;有一个优化器Catalyst&＃xff0c;具体介绍看后面参考的文章。

上面说到的这些RDD的弊端&＃xff0c;有一部分就可以使用mapPartitions进行优化&＃xff0c;mapPartitions可以同时替代rdd.map,rdd.filter,rdd.flatMap的作用&＃xff0c;所以在长操作中&＃xff0c;可以在mapPartitons中将RDD大量的操作写在一起&＃xff0c;避免产生大量的中间rdd对象&＃xff0c;另外是mapPartitions在一个partition中可以复用可变类型&＃xff0c;这也能够避免频繁的创建新对象。使用mapPartitions的弊端就是牺牲了代码的易读性。

在大数据分布式系统中&＃xff0c;大量数据的移动对性能的影响也是巨大的。基于这个思想&＃xff0c;在两个RDD进行join操作的时候&＃xff0c;如果其中一个RDD相对小很多&＃xff0c;可以将小的RDD进行collect操作然后设置为broadcast变量&＃xff0c;这样做之后&＃xff0c;另一个RDD就可以使用map操作进行join&＃xff0c;这样能够有效的减少相对大很多的那个RDD的数据移动。

这个就是谓词下推&＃xff0c;这个很显然&＃xff0c;filter之后再join&＃xff0c;shuffle的数据量会减少&＃xff0c;这里提一点是spark-sql的优化器已经对这部分有优化了&＃xff0c;不需要用户显示的操作&＃xff0c;个人实现rdd的计算的时候需要注意这个。

这一部分在另一篇文章《spark partitioner使用技巧 》有详细介绍&＃xff0c;这里不说了。

这个操作在Map-Reduce中也有&＃xff0c;这里举个例子&＃xff1a;rdd.groupByKey().mapValue(_.sum)比rdd.reduceByKey的效率低&＃xff0c;原因如下两幅图所示(网上盗来的&＃xff0c;侵删)

上下两幅图的区别就是上面那幅有combineByKey的过程减少了shuffle的数据量&＃xff0c;下面的没有。combineByKey是key-value型rdd自带的API&＃xff0c;可以直接使用。

rdd.cache()和rdd.persist(Storage.MEMORY_ONLY)是等价的&＃xff0c;在内存不足的时候rdd.cache()的数据会丢失&＃xff0c;再次使用的时候会重算&＃xff0c;而rdd.persist(StorageLevel.MEMORY_AND_DISK_SER)在内存不足的时候会存储在磁盘&＃xff0c;避免重算&＃xff0c;只是消耗点IO时间。

在spark结合hbase的使用中&＃xff0c;spark和hbase最好搭建在同一个集群上上&＃xff0c;或者spark的集群节点能够覆盖hbase的所有节点。hbase中的数据存储在HFile中&＃xff0c;通常单个HFile都会比较大&＃xff0c;另外Spark在读取Hbase的数据的时候&＃xff0c;不是按照一个HFile对应一个RDD的分区&＃xff0c;而是一个region对应一个RDD分区。所以在Spark读取Hbase的数据时&＃xff0c;通常单个RDD都会比较大&＃xff0c;如果不是搭建在同一个集群&＃xff0c;数据移动会耗费很多的时间。

这个参数用来设置Driver的内存。在Spark程序中&＃xff0c;SparkContext&＃xff0c;DAGScheduler都是运行在Driver端的。对应rdd的Stage切分也是在Driver端运行&＃xff0c;如果用户自己写的程序有过多的步骤&＃xff0c;切分出过多的Stage&＃xff0c;这部分信息消耗的是Driver的内存&＃xff0c;这个时候就需要调大Driver的内存。

这个参数在内存吃紧的时候&＃xff0c;又需要persist数据有良好的性能&＃xff0c;就可以设置这个参数为true&＃xff0c;这样在使用persist(StorageLevel.MEMORY_ONLY_SER)的时候&＃xff0c;就能够压缩内存中的rdd数据。减少内存消耗&＃xff0c;就是在使用的时候会占用CPU的解压时间。

建议设置为 org.apache.spark.serializer.KryoSerializer&＃xff0c;因为KryoSerializer比JavaSerializer快&＃xff0c;但是有可能会有些Object会序列化失败&＃xff0c;这个时候就需要显示的对序列化失败的类进行KryoSerializer的注册&＃xff0c;这个时候要配置spark.kryo.registrator参数或者使用参照如下代码&＃xff1a;

这个参数设置内存表示 Executor内存中 storage/(storage&＃43;execution)&＃xff0c;虽然spark-1.6.0&＃43;的版本内存storage和execution的内存已经是可以互相借用的了&＃xff0c;但是借用和赎回也是需要消耗性能的&＃xff0c;所以如果明知道程序中storage是多是少就可以调节一下这个参数。

spark中有4中本地化执行level&＃xff0c;PROCESS_LOCAL->NODE_LOCAL->RACK_LOCAL->ANY,一个task执行完&＃xff0c;等待spark.locality.wait时间如果&＃xff0c;第一次等待PROCESS的Task到达&＃xff0c;如果没有&＃xff0c;等待任务的等级下调到NODE再等待spark.locality.wait时间&＃xff0c;依次类推&＃xff0c;直到ANY。分布式系统是否能够很好的执行本地文件对性能的影响也是很大的。如果RDD的每个分区数据比较多&＃xff0c;每个分区处理时间过长&＃xff0c;就应该把 spark.locality.wait 适当调大一点&＃xff0c;让Task能够有更多的时间等待本地数据。特别是在使用persist或者cache后&＃xff0c;这两个操作过后&＃xff0c;在本地机器调用内存中保存的数据效率会很高&＃xff0c;但是如果需要跨机器传输内存中的数据&＃xff0c;效率就会很低。

一个大的集群中&＃xff0c;每个节点的性能会有差异&＃xff0c;spark.speculation这个参数表示空闲的资源节点会不会尝试执行还在运行&＃xff0c;并且运行时间过长的Task&＃xff0c;避免单个节点运行速度过慢导致整个任务卡在一个节点上。这个参数最好设置为true。与之相配合可以一起设置的参数有spark.speculation.×开头的参数。参考中有文章详细说明这个参数。

以后有遇到新的内容再补充。

1. http://www.jianshu.com/p/c0181667daa0

2. http://www.csdn.net/article/2015-06-18/2824958

3. https://chenzhongpu.gitbooks.io/bigdatanotes/content/SparkSQLOptimizer/index.html

4. http://book.51cto.com/art/201409/453045.htm