Dlaczego zadania Sparka kończą się niepowodzeniem z org.apache.spark.shuffle.MetadataFetchFailedException: Brakuje lokalizacji wyjściowej dla losowania 0 w trybie spekulacji?

Question

Używam pracy Spark w trybie spekulacji. Mam około 500 zadań i około 500 plików o 1 GB skompresowanych gz. Ciągle dostaję się do każdej pracy, dla 1-2 zadań, załączonego błędu, gdzie powtarza się po dziesiątki razy (uniemożliwiając wykonanie zadania).

org.apache.spark.shuffle.MetadataFetchFailedException: Brakujące miejsca odbioru dla Shuffle 0

jakiś pomysł co jest znaczenie tego problemu i jak go pokonać?

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

Stało się to, gdy dałem więcej pamięci węzłowi roboczemu niż ma. Ponieważ nie miała ona zamiany, iskra uległa awarii podczas próby przechowywania obiektów do przetasowania bez pozostawienia pamięci.

Rozwiązaniem było dodanie wymiany lub skonfigurowanie robota/executora do użycia mniej pamięci, a także użycie poziomu pamięci MEMORY_AND_DISK przez kilka powtórzeń.

Answer 2

Mieliśmy podobny błąd ze Spark, ale nie jestem pewien, czy jest to związane z twoim problemem.

Użyliśmy JavaPairRDD.repartitionAndSortWithinPartitions na 100 GB danych i nie powiodło się podobnie do Twojej aplikacji. Następnie przyjrzeliśmy się dziennikom przędzy na poszczególnych węzłach i dowiedzieliśmy się, że mamy problem z brakiem pamięci, więc Przędza przerwała egzekucję. Naszym rozwiązaniem było zmienić/dodać spark.shuffle.memoryFraction 0 w .../spark/conf/spark-defaults.conf. To pozwoliło nam obsługiwać znacznie większą (ale niestety nieskończoną) ilość danych w ten sposób.

Answer 3

Mam ten sam problem w moim 3 klastrze YARN. Zmieniłem pamięć RAM, ale problem nadal występował. Wreszcie ujrzałem następujące komunikaty w dziennikach:

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 

i po tym, było to komunikat:

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

I zmodyfikowane właściwości w zapłonowej-defaults.conf następująco:

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

To wszystko! Moja praca zakończyła się pomyślnie po tym.

Answer 4

W moim przypadku (samodzielny klaster) wyjątek został zgłoszony, ponieważ system plików niektórych niewolników Spark został wypełniony w 100%. Usunięcie wszystkiego z folderów niewolników spark/work rozwiązało problem.

Answer 5

Ten problem dotyczy około pamięci. Mijasz pamięć, która nie jest dostępna. W poleceniu iskr submit są parametry dotyczące pamięci i wykonania. Są to parametry są

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

swoja pamięć sterownika max n/2 (n = całkowite pamięć węzła.)

Daj kierowca rdzenie n/1000 (n = pamięć sterownika (w MB))

Daj Num-eXecutors n (n = liczba węzłów masz)

Jeśli nie dadzą odpowiednie parametry komenda iskry, następnie będzie powolna reakcja lub spowoduje wyjątek.