Spark: nicht genügend Speicher beim Senden von Objekten

Question

Ich habe versucht, eine nicht so große Karte (~ 70 MB, wenn in HDFS als Textdatei gespeichert) zu senden, und ich bekam keine Speicherfehler. Ich habe versucht, den Treiberspeicher auf 11G und Executor-Speicher auf 11G zu erhöhen, und immer noch den gleichen Fehler. Die memory.fraction wird auf 0.3 gesetzt und es werden auch nicht viele Daten (weniger als 1G) zwischengespeichert.

Wenn die Karte nur etwa 2 MB ist, gibt es kein Problem. Ich frage mich, ob es beim Senden von Objekten eine Größenbeschränkung gibt. Wie kann ich dieses Problem mit der größeren Karte lösen? Vielen Dank!

Exception in thread "main" java.lang.OutOfMemoryError: Java heap space 
    at java.util.IdentityHashMap.resize(IdentityHashMap.java:469) 
    at java.util.IdentityHashMap.put(IdentityHashMap.java:445) 
    at org.apache.spark.util.SizeEstimator$SearchState.enqueue(SizeEstimator.scala:159) 
    at org.apache.spark.util.SizeEstimator$.visitArray(SizeEstimator.scala:229) 
    at org.apache.spark.util.SizeEstimator$.visitSingleObject(SizeEstimator.scala:194) 
    at org.apache.spark.util.SizeEstimator$.org$apache$spark$util$SizeEstimator$$estimate(SizeEstimator.scala:186) 
    at org.apache.spark.util.SizeEstimator$.estimate(SizeEstimator.scala:54) 
    at org.apache.spark.util.collection.SizeTracker$class.takeSample(SizeTracker.scala:78) 
    at org.apache.spark.util.collection.SizeTracker$class.afterUpdate(SizeTracker.scala:70) 
    at org.apache.spark.util.collection.SizeTrackingVector.$plus$eq(SizeTrackingVector.scala:31) 
    at org.apache.spark.storage.MemoryStore.unrollSafely(MemoryStore.scala:278) 
    at org.apache.spark.storage.MemoryStore.putIterator(MemoryStore.scala:165) 
    at org.apache.spark.storage.MemoryStore.putIterator(MemoryStore.scala:143) 
    at org.apache.spark.storage.BlockManager.doPut(BlockManager.scala:801) 
    at org.apache.spark.storage.BlockManager.putIterator(BlockManager.scala:648) 
    at org.apache.spark.storage.BlockManager.putSingle(BlockManager.scala:1006) 
    at org.apache.spark.broadcast.TorrentBroadcast.writeBlocks(TorrentBroadcast.scala:99) 
    at org.apache.spark.broadcast.TorrentBroadcast.<init>(TorrentBroadcast.scala:85) 
    at org.apache.spark.broadcast.TorrentBroadcastFactory.newBroadcast(TorrentBroadcastFactory.scala:34) 
    at org.apache.spark.broadcast.BroadcastManager.newBroadcast(BroadcastManager.scala:63) 
    at org.apache.spark.SparkContext.broadcast(SparkContext.scala:1327) 

Edit: Weitere Informationen hinzufügen nach den Kommentaren:

• ich Funken einreichen die kompilierte JAR-Datei im Client-Modus zu senden. Funken 1.5.0
• spark.yarn.executor.memoryOverhead 600
• Set ("spark.kryoserializer.buffer.max", "256M")
• Set ("spark.speculation", "true")
• Set ("spark.storage.memoryFraction", "0,3")
• Set ("spark.driver.memory", "15G")
• Set ("spark.executor.memory", "11G")
• Ich versuchte Set ("spar.sql.tungsten.enabled", "false") und es hilft nicht.
• Die Master-Maschine hat 60G Speicher. Rund 30G wird für Spark/Yarn verwendet. Ich bin nicht sicher, wie viel Heap-Größe für meinen Job ist, aber es gibt nicht viel andere Prozesse zur gleichen Zeit. Vor allem die Karte ist nur um 70MB.

Einige Code auf die Ausstrahlung bezogen werden:

val mappingAllLocal: Map[String, Int] = mappingAll.rdd.map(r => (r.getAs[String](0), r.getAs[Int](1))).collectAsMap().toMap 
// I can use the above mappingAll to HDFS, and it's around 70MB 
val mappingAllBrd = sc.broadcast(mappingAllLocal) // <-- this is where the out of memory happens 

Answer 1

set("spark.driver.memory", "15G") Verwendung hat keine Auswirkung auf den Client-Modus. Sie müssen beim Übergeben der Anwendung den Befehlszeilenparameter --conf="spark.driver.memory=15G" verwenden, um die Heap-Größe des Treibers zu erhöhen.

Answer 2

Sie versuchen, die JVM-Heap-Größe erhöhen:

-Xmx2g : max size of 2Go 
-Xms2g : initial size of 2Go (default size is 256mo)