标签:
本文目的
?
最近使用spark处理较大的数据时,遇到了分区2G限制的问题(ken)。找到了解决方法,并且在网上收集了一些资料,记录在这里,作为备忘。
?
问题现象
?
遇到这个问题时,spark日志会报如下的日志,
片段1
15/04/16 14:13:03 WARN scheduler.TaskSetManager: Lost task 19.0 in stage 6.0 (TID 120, 10.215.149.47): java.lang.IllegalArgumentException: Size exceeds Integer.MAX_VALUE |
?
片段2
15/04/16 14:19:45 INFO scheduler.TaskSetManager: Starting task 20.2 in stage 6.0 (TID 146, 10.196.151.213, PROCESS_LOCAL, 1666 bytes) 15/04/16 14:19:45 INFO scheduler.TaskSetManager: Lost task 20.2 in stage 6.0 (TID 146) on executor 10.196.151.213: java.lang.IllegalArgumentException (Size exceeds Integer.MAX_VALUE) [duplicate 1] 15/04/16 14:19:45 INFO scheduler.TaskSetManager: Starting task 20.3 in stage 6.0 (TID 147, 10.196.151.213, PROCESS_LOCAL, 1666 bytes) 15/04/16 14:19:45 INFO scheduler.TaskSetManager: Lost task 20.3 in stage 6.0 (TID 147) on executor 10.196.151.213: java.lang.IllegalArgumentException (Size exceeds Integer.MAX_VALUE) [duplicate 2] 15/04/16 14:19:45 ERROR scheduler.TaskSetManager: Task 20 in stage 6.0 failed 4 times; aborting job 15/04/16 14:19:45 INFO cluster.YarnClusterScheduler: Cancelling stage 6 15/04/16 14:19:45 INFO cluster.YarnClusterScheduler: Stage 6 was cancelled 15/04/16 14:19:45 INFO scheduler.DAGScheduler: Job 6 failed: collectAsMap at DecisionTree.scala:653, took 239.760845 s 15/04/16 14:19:45 ERROR yarn.ApplicationMaster: User class threw exception: Job aborted due to stage failure: Task 20 in stage 6.0 failed 4 times, most recent failure: Lost task 20.3 in stage 6.0 (TID 147, 10.196.151.213): java.lang.IllegalArgumentException: Size exceeds Integer.MAX_VALUE at sun.nio.ch.FileChannelImpl.map(FileChannelImpl.java:828) |
?
注意红色高亮部分,异常就是某个partition的数据量超过了Integer.MAX_VALUE(2147483647 = 2GB)。
?
解决方法
?
手动设置RDD的分区数量。当前使用的Spark默认RDD分区是18个,后来手动设置为1000个,上面这个问题就迎刃而解了。可以在RDD加载后,使用RDD.repartition(numPart:Int)函数重新设置分区数量。
?
为什么2G限制
?
目前spark社区对这个限制有很多讨(tu)论(cao),spark官方团队已经注意到了这个问题,但是直到1.2版本,这个问题还是没有解决。因为牵涉到整个RDD的实现框架,所以改进成本相当大!
?
下面是一些相关的资料,有兴趣的读者可以进一步的阅读:
?
个人思(yu)考(jian)
?
这个限制有一定合理性。因为RDD中partition的操作是并发执行的,如果partition量过少,导致并发数过少,会限制计算效率。所以,基于这个限制,spark应用程序开发者会主动扩大partition数量,也就是加大并发量,最终提高计算性能。
?
以上只是一些个能思考,如果不正确,还请拍砖。
标签:
原文地址:http://www.cnblogs.com/bourneli/p/4456109.html