火花错误：java.lang.IllegalArgumentException：大小超过 Integer.MAX_VALUE答案

【问题标题】：spark error:java.lang.IllegalArgumentException: Size exceeds Integer.MAX_VALUE火花错误：java.lang.IllegalArgumentException：大小超过 Integer.MAX_VALUE
【发布时间】：2018-09-23 10:05:01
【问题描述】：

我尝试计算负样本的数量，如下：

val numNegatives = dataSet.filter(col("label") < 0.5).count

但我收到了 Size 超过 Integer.MAX_VALUE 错误：

java.lang.IllegalArgumentException: Size exceeds Integer.MAX_VALUE
    at sun.nio.ch.FileChannelImpl.map(FileChannelImpl.java:869)
    at org.apache.spark.storage.DiskStore$$anonfun$getBytes$2.apply(DiskStore.scala:127)
    at org.apache.spark.storage.DiskStore$$anonfun$getBytes$2.apply(DiskStore.scala:115)
    at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1239)
    at org.apache.spark.storage.DiskStore.getBytes(DiskStore.scala:129)
    at org.apache.spark.storage.DiskStore.getBytes(DiskStore.scala:136)
    at org.apache.spark.storage.BlockManager.doGetLocal(BlockManager.scala:512)
    at org.apache.spark.storage.BlockManager.getLocal(BlockManager.scala:427)
    at org.apache.spark.storage.BlockManager.get(BlockManager.scala:636)
    at org.apache.spark.CacheManager.getOrCompute(CacheManager.scala:44)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:268)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:306)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:270)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:306)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:270)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:306)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:270)
    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:89)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:227)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

一些解决方案建议添加分区号，所以我将上面的代码更新为：

val data = dataSet.repartition(5000).cache()
val numNegatives = data.filter(col("label") < 0.5).count

但它报告了同样的错误！这让我困惑了几天。谁能帮我？谢谢。

【问题讨论】：

我认为您的数据集的一列包含不适合的数字。您如何获得该数据集？
我用spark-hive sql得到了数据集：select id,cast(label as double) label, ... from table。看来每一列都应该互相适应。

标签： scala apache-spark machine-learning

【解决方案1】：

她的问题是实现后的 ShuffleRDD 块大小大于 2GB。 Spark 有这个limitation。您需要更改默认设置为 200 的 spark.sql.shuffle.partitions 参数。

此外，您可能需要增加数据集的分区数量。重新分区并先保存，然后读取新的数据集并执行操作。

spark.sql("SET spark.sql.shuffle.partitions = 10000")
dataset.repartition(10000).write.parquet("/path/to/hdfs")
val newDataset = spark.read.parquet("/path/to/hdfs")  
newDatase.filter(...).count

如果您想使用 Hive Table 也可以选择

spark.sql("SET spark.sql.shuffle.partitions = 10000")
dataset.repartition(10000).asveAsTable("newTableName")
val newDataset = spark.table("newTableName")  
newDatase.filter(...).count

【讨论】：

我试过这样：val data = dataSet.repartition(5000).cache() val numNegatives = data.filter(col("label")
不要缓存。请增加分区..设为10000，重新分区后先将数据保存在HDfs或s3上。然后加载它并进行任何操作。另外，正如我所建议的，请将 spark.sql.shuffle.partition 增加到 10000。我已经多次遇到这个问题，这对我有用
hdfs文件是hive sql插入的，只有一个文件，所以文件大小为2.13G。在 hive sql 中我可以重新分区 hdfs 文件吗？
我已经更新了答案，其中包含解决问题所需的确切步骤。

【解决方案2】：

尝试在过滤器之前重新分区：

val numNegatives = dataSet.repartition(1000).filter(col("label")

过滤器使用原始数据集分区执行并重新分区结果。您需要为过滤器设置更小的分区。

【讨论】：