Warum hat die Spark RDD-Partition 2 GB für HDFS?

Question

Ich habe einen Fehler bei der Verwendung von mllib RandomForest, um Daten zu trainieren. Da meine Datenmenge sehr groß ist und die Standardpartition relativ klein ist. so wird eine Ausnahme ausgelöst darauf hinweist, dass "Größe überschreitet Integer.MAX_VALUE", die Orignal-Stack-Trace, wie folgend,

15/04/16 14:13:03 WARN scheduler.TaskSetManager: Verlorene Aufgabe 19.0 in Stufe 6.0 (TID 120, 10.215.149.47): java.lang.IllegalArgumentException: Größe überschreitet Integer.MAX_VALUE
bei sun.nio.ch.FileChannelImpl.map (FileChannelImpl.java:828) bei org.apache.spark.storage.DiskStore .getBytes (DiskStore.scala: 123) bei org.apache.spark.storage.DiskStore.getBytes (DiskStore.scala: 132) bei org.apache.spark.storage.BlockManager.doGetLocal (BlockManager.scala: 517) um org.apache.spark.storage.BlockManager.getLocal (BlockManager.scala: 432) bei org.apache.spark.storage.BlockManager.get (BlockManager.scala: 618) unter org.apache.spark.CacheManager. putInBlockManager (CacheManager.scala: 146) bei org.apache.spark.CacheManager.getOrCompute (CacheManager.scala: 70)

Die Integer.MAX_SIZE ist 2GB, scheint es, dass einige Partition aus dem Speicher. Also repartiton ich meine rdd-Partition auf 1000, so dass jede Partition weit weniger Daten wie zuvor speichern konnte. Endlich ist das Problem gelöst !!!

Also, meine Frage ist: Warum Partition Größe hat die 2G Grenze? Es scheint, dass es keine Konfiguration Set für die Grenze in der Funke

Answer 1

Die grundlegende Abstraktion für Blöcke in Spark ist ein ByteBuffer, die leider eine Grenze von Integer.MAX_VALUE (~ 2GB) hat.

Es ist ein critical issue, der die Verwendung von Funken mit sehr großen Datensätzen verhindert. Das Erhöhen der Anzahl der Partitionen kann es auflösen (wie im Fall von OP), ist aber nicht immer möglich, z. B. wenn eine große Kette von Transformationen Teil von Daten erhöhen kann (flatMap usw.) oder wenn Daten schief sind.

Die vorgeschlagene Lösung besteht darin, eine Abstraktion wie LargeByteBuffer zu erstellen, die eine Liste von Byebuffern für einen Block unterstützen kann. Dies wirkt sich auf die gesamte Spark-Architektur aus und ist daher noch lange nicht gelöst.