Perché i processi di Spark falliscono con org.apache.spark.shuffle.MetadataFetchFailedException: Manca un percorso di output per lo shuffle 0 in modalità speculazione?

Question

Sto eseguendo un lavoro Spark con una modalità di speculazione. Ho circa 500 attività e circa 500 file di 1 GB gz compressi. Continuo a ottenere in ogni lavoro, per 1-2 compiti, l'errore allegato dove si replica dopo decine di volte (impedendo il completamento del lavoro).

org.apache.spark.shuffle.MetadataFetchFailedException: Manca una posizione di uscita per il riordino 0

Qualsiasi idea di qual è il significato del problema e come superarla?

org.apache.spark.shuffle.MetadataFetchFailedException: Missing an output location for shuffle 0 
    at org.apache.spark.MapOutputTracker$$anonfun$org$apache$spark$MapOutputTracker$$convertMapStatuses$1.apply(MapOutputTracker.scala:384) 
    at org.apache.spark.MapOutputTracker$$anonfun$org$apache$spark$MapOutputTracker$$convertMapStatuses$1.apply(MapOutputTracker.scala:381) 
    at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244) 
    at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244) 
    at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33) 
    at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:108) 
    at scala.collection.TraversableLike$class.map(TraversableLike.scala:244) 
    at scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:108) 
    at org.apache.spark.MapOutputTracker$.org$apache$spark$MapOutputTracker$$convertMapStatuses(MapOutputTracker.scala:380) 
    at org.apache.spark.MapOutputTracker.getServerStatuses(MapOutputTracker.scala:176) 
    at org.apache.spark.shuffle.hash.BlockStoreShuffleFetcher$.fetch(BlockStoreShuffleFetcher.scala:42) 
    at org.apache.spark.shuffle.hash.HashShuffleReader.read(HashShuffleReader.scala:40) 
    at org.apache.spark.rdd.ShuffledRDD.compute(ShuffledRDD.scala:92) 
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263) 
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:230) 
    at org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31) 
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263) 
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:230) 
    at org.apache.spark.rdd.FlatMappedRDD.compute(FlatMappedRDD.scala:33) 
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263) 
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:230) 
    at org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31) 
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263) 
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:230) 
    at org.apache.spark.rdd.MappedRDD.compute(MappedRDD.scala:31) 
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:263) 
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:230) 
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61) 
    at org.apache.spark.scheduler.Task.run(Task.scala:56) 
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196) 
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) 
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) 
    at java.lang.Thread.run(Thread.java:722) 

Answer 1

Questo è successo a me quando ho dato più memoria al nodo di lavoro di quello che ha. Dato che non è stato scambiato, la scintilla si è interrotta durante il tentativo di memorizzare oggetti per mischiare senza più memoria.

La soluzione era aggiungere o scambiare o configurare il worker/executor per utilizzare meno memoria oltre a utilizzare il livello di memoria MEMORY_AND_DISK per diversi persistenti.

Answer 2

Abbiamo avuto un errore simile con Spark, ma non sono sicuro che sia correlato al problema.

Abbiamo utilizzato JavaPairRDD.repartitionAndSortWithinPartitions su 100 GB di dati e non ha funzionato in modo simile alla tua app. Poi abbiamo esaminato i log del filato sui nodi specifici e abbiamo scoperto che abbiamo qualche tipo di problema di memoria esaurita, quindi il filato ha interrotto l'esecuzione. La nostra soluzione era di modificare/aggiungere spark.shuffle.memoryFraction 0 in .../spark/conf/spark-defaults.conf. Questo ci ha permesso di gestire una quantità di dati molto più grande (ma purtroppo non infinita) in questo modo.

Answer 3

Ho ricevuto lo stesso problema sul mio cluster YARN a 3 macchine. Ho continuato a cambiare la RAM ma il problema persisteva. Finalmente ho visto i seguenti messaggi nei registri:

17/02/20 13:11:02 WARN spark.HeartbeatReceiver: Removing executor 2 with no recent heartbeats: 1006275 ms exceeds timeout 1000000 ms 
17/02/20 13:11:02 ERROR cluster.YarnScheduler: Lost executor 2 on 1worker.com: Executor heartbeat timed out after 1006275 ms 

e dopo questo, c'era questo messaggio:

org.apache.spark.shuffle.MetadataFetchFailedException: Missing an output location for shuffle 67 

ho modificato le proprietà a scintilla defaults.conf come segue:

spark.yarn.scheduler.heartbeat.interval-ms 7200000 
spark.executor.heartbeatInterval 7200000 
spark.network.timeout 7200000 

Questo è tutto! Il mio lavoro è stato completato con successo dopo questo.

Answer 4

Nel mio caso (cluster autonomo) è stata generata un'eccezione perché il file system di alcuni slave Spark è stato riempito al 100%. L'eliminazione di tutto nelle cartelle spark/work degli slave ha risolto il problema.

Answer 5

Questo problema riguarda la memoria . Stai passando la memoria che non è disponibile. Nel comando spark invia ci sono i parametri relativi alla memoria e all'esecuzione. Questi parametri sono sono

--driver-memory 1200M 
--driver-cores 1 
--num-executors 1 

Lascia memoria conducente max n/2 (n = memoria totale del nodo.)

Dare conducenti-core n/1000 (n = memoria del driver (in MB))

Fai num-esecutori n (n = numero di nodi che avete)

Se non darà parametri corretti in il comando di invio scintilla quindi ci sarà una risposta lenta o causerà un'eccezione.