为什么 Spark RDD 分区对 HDFS 有 2GB 的限制？

Question

使用 mllib RandomForest 训练数据时出现错误。由于我的数据集很大，而默认分区相对较小。所以抛出一个异常，表明“Size超过Integer.MAX_VALUE”，原始堆栈跟踪如下，

15/04/16 14:13:03 WARN scheduler.TaskSetManager: Lost task 19.0 in 阶段 6.0（TID 120、10.215.149.47）： java.lang.IllegalArgumentException：大小超过 Integer.MAX_VALUE
在 sun.nio.ch.FileChannelImpl.map(FileChannelImpl.java:828) 在 org.apache.spark.storage.DiskStore.getBytes(DiskStore.scala:123) 在 org.apache.spark.storage.DiskStore.getBytes(DiskStore.scala:132) 在 org.apache.spark.storage.BlockManager.doGetLocal(BlockManager.scala:517) 在 org.apache.spark.storage.BlockManager.getLocal(BlockManager.scala:432) 在 org.apache.spark.storage.BlockManager.get(BlockManager.scala:618) 在 org.apache.spark.CacheManager.putInBlockManager(CacheManager.scala:146) 在 org.apache.spark.CacheManager.getOrCompute(CacheManager.scala:70)

Integer.MAX_SIZE 为 2GB，似乎是某个分区内存不足。所以我将我的 rdd 分区重新分区为 1000，以便每个分区可以像以前一样保存更少的数据。终于，问题解决了！！！

所以，我的问题是： 为什么分区大小有2G限制？ spark中的limit好像没有设置配置

Answer 1

spark 中块的基本抽象是ByteBuffer，不幸的是它有一个 Integer.MAX_VALUE (~2GB) 的限制。

这是一个critical issue，它可以防止在非常大的数据集上使用 spark。 增加分区的数量可以解决它（就像在 OP 的情况下一样），但并不总是可行的，例如，当存在大的转换链时，其中一部分可以增加数据（flatMap 等）或在数据倾斜的情况下。

提出的解决方案是提出像LargeByteBuffer 这样的抽象，它可以支持块的字节缓冲区列表。这会影响整个 Spark 架构，因此在很长一段时间内仍未解决。

Answer 2

问题在于，当使用 Casandra、HBase 或 Accumulo 等数据存储时，块大小取决于数据存储拆分（可能超过 10 gig）。从这些数据存储加载数据时，您必须立即使用 1000 个分区重新分区，这样您就可以在不超出 2gig 限制的情况下操作数据。

大多数使用 spark 的人并没有真正使用大数据；对他们来说，如果 excel 可以容纳更大，或者 tableau 对他们来说是大数据；大多数是数据科学家，他们使用质量数据或使用足够小的样本量来处理极限。

当处理大量数据时，我不得不回到 mapreduce 并且只在数据被清理后才使用 spark。不幸的是，大多数 Spark 社区都没有兴趣解决这个问题。

一个简单的解决方案是创建一个抽象并使用 bytearray 作为默认值；但是，允许使用 64 位数据指针重载 spark 作业来处理大型作业。

Answer 3

Spark 2.4.0 release 通过将块数据复制为流来消除此限制。详情请见Spark-24926。