spark job运行参数优化-阿里云开发者社区

开发者社区> 力君> 正文

spark job运行参数优化

简介:
+关注继续查看

一、问题

      使用spark join两张表(5000w*500w)总是出错,报的异常显示是在shuffle阶段。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
14/11/27 12:05:49 ERROR storage.DiskBlockObjectWriter: Uncaught exception while reverting partial writes to file /hadoop/application_1415632483774_448143/spark-local-20141127115224-9ca8/04/shuffle_1_1562_27
java.io.FileNotFoundException: /hadoop/application_1415632483774_448143/spark-local-20141127115224-9ca8/04/shuffle_1_1562_27 (No such file or directory)
        at java.io.FileOutputStream.open(Native Method)
        at java.io.FileOutputStream.<init>(FileOutputStream.java:212)
        at org.apache.spark.storage.DiskBlockObjectWriter.revertPartialWritesAndClose(BlockObjectWriter.scala:178)
        at org.apache.spark.shuffle.hash.HashShuffleWriter$$anonfun$revertWrites$1.apply(HashShuffleWriter.scala:118)
        at org.apache.spark.shuffle.hash.HashShuffleWriter$$anonfun$revertWrites$1.apply(HashShuffleWriter.scala:117)
        at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
        at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:108)
        at org.apache.spark.shuffle.hash.HashShuffleWriter.revertWrites(HashShuffleWriter.scala:117)
        at org.apache.spark.shuffle.hash.HashShuffleWriter.stop(HashShuffleWriter.scala:89)
        at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73)
        at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
        at org.apache.spark.scheduler.Task.run(Task.scala:54)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:177)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
        at java.lang.Thread.run(Thread.java:724)

     

    出问题的代码块(scala)

1 val cRdd = iRdd.leftOuterJoin(label).map {
2      case (id, (iMap, Some(set))) => (id, (iMap, set))
3      case (id, (iMap, None)) => (id, (iMap, new HashSet[Int]()))
4    }.persist(StorageLevel.MEMORY_AND_DISK)

 

二、问题分析与解决

     一般spark job很多问题都是来源于系统资源不够用,通过监控日志等判断是内存资源占用过高导致的问题,因此尝试通过配置参数的方法来解决。

1)--conf spark.akka.frameSize=100

     此参数控制Spark中通信消息的最大容量 (如task的输出结果),默认为10M。当处理大数据时,task的输出可能会大于这个值,需要根据实际数据设置一个更高的值。尝试将此参数设置成100M后,问题未能解决。

2)--conf spark.shuffle.manager=SORT

     Spark默认的shuffle采用Hash模式,在HASH模式下,每一次shuffle会生成M*R的数量的文件(M指的是Map的数目,R指的是Reduce的数目),而当Map和Reduce的数目开得较大时,会产生相当规模的文件,与此同时带来了大量的内存开销。

     为了降低系统资源,可以采用Sort模式,Sort模式只产生M数量的文件。具体可以参考:Sort-based Shuffle之初体验

     在我们的应用场景下,采用Sort模式后,shuffle时间比之前增大了1/3,但是问题依旧未解决。

3)--conf spark.yarn.executor.memoryOverhead=4096

     executor堆外内存设置。起初是1024M,未能跑过,后改为4096M,Job就能跑通,原因是程序使用了大量的堆外内存。

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
一次job的耗时优化经历
说起这次优化,有点奇葩,从耗时12小时到1分钟,优化经历了三次,和自己纠结了N次,看看结果还算安慰。
2046 0
MySQL SSD 参数优化
SSD硬盘:慢查询日志可以设置0.5秒,如果超过0.5秒。0.5秒在SSD上最少走了50个IO,就有可能没有用到索引。0.5秒还是有点问题:如果从8000W中找一条记录,如果加上order 等计算耗时,比较小。
3400 0
Spark 资源和数据并行度优化分析2 | 学习笔记
快速学习 Spark 资源和数据并行度优化分析2
20 0
参数(parameter)和属性(Attribute)的区别
参数(parameter)和属性(Attribute)的区别 区别:  来源不同:    参数(parameter)是从客户端(浏览器)中由用户提供的,若是GET方法是从URL中  提供的,若是POST方法是从请求体(request body)中提供的;    属性(attribute)是服务器端的组件(JSP或者Servlet)利用requst.
866 0
基于英特尔® 优化分析包(OAP)的 Spark 性能优化方案
Spark SQL 作为 Spark 用来处理结构化数据的一个基本模块,已经成为多数企业构建大数据应用的重要选择。但是,在大规模连接(Join)、聚合(Aggregate)等工作负载下,Spark 性能会面临稳定性和性能方面的挑战。
483 0
+关注
力君
阿里巴巴RDC工程师 邮箱:06lksyzlj@gmail.com
21
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
《2021云上架构与运维峰会演讲合集》
立即下载
《零基础CSS入门教程》
立即下载
《零基础HTML入门教程》
立即下载