RDD分區2GB限制

本文目的

 

最近使用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 at sun.nio.ch.FileChannelImpl.map(FileChannelImpl.java:828) at org.apache.spark.storage.DiskStore.getBytes(DiskStore.scala:123) at org.apache.spark.storage.DiskStore.getBytes(DiskStore.scala:132) at org.apache.spark.storage.BlockManager.doGetLocal(BlockManager.scala:517) at org.apache.spark.storage.BlockManager.getLocal(BlockManager.scala:432) at org.apache.spark.storage.BlockManager.get(BlockManager.scala:618) at org.apache.spark.CacheManager.putInBlockManager(CacheManager.scala:146) at org.apache.spark.CacheManager.getOrCompute(CacheManager.scala:70)

 

片斷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的實現框架，因此改進成本至關大!

 

下面是一些相關的資料，有興趣的讀者能夠進一步的閱讀:

• 2GB limit in spark for blocks
• create LargeByteBuffer abstraction for eliminating 2GB limit on blocks
• Why does Spark RDD partition has 2GB limit for HDFS
• 拋異常的java代碼:FileChannelImpl.java

 

我的思（yu）考（jian）

 

這個限制有必定合理性。由於RDD中partition的操做是併發執行的，若是partition量過少，致使併發數過少，會限制計算效率。因此，基於這個限制，spark應用程序開發者會主動擴大partition數量，也就是加大併發量，最終提升計算性能。

 

以上只是一些個能思考，若是不正確，還請拍磚。