Warum schlagen Spark-Jobs mit org.apache.spark.shuffle.MetadataFetchFailedException fehl: Fehlender Ausgabeort für Shuffle 0 im Spekulationsmodus?

83

Ich führe einen Spark-Job mit in einem Spekulationsmodus aus. Ich habe ungefähr 500 Aufgaben und ungefähr 500 Dateien mit 1 GB gz komprimiert. Ich bekomme in jedem Job für 1-2 Aufgaben den angehängten Fehler, bei dem er dutzende Male wiederholt wird (wodurch verhindert wird, dass der Job abgeschlossen wird).

org.apache.spark.shuffle.MetadataFetchFailedException: Fehlender Ausgabeort für Shuffle 0

Irgendeine Idee, was das Problem bedeutet und wie man es überwinden kann?

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)
apache-spark Dotan
quelle
1
Haben Sie LostExecutorINFO-Nachrichten gesehen? Können Sie die Executors-Seite der Web-Benutzeroberfläche überprüfen und sehen, wie sich Executors verhalten, insb. GC-weise?
Jacek Laskowski

Antworten:

50

Dies ist mir passiert, als ich dem Worker-Knoten mehr Speicher gegeben habe als er hat. Da es keinen Austausch gab, stürzte der Funke ab, als versucht wurde, Objekte zum Mischen zu speichern, ohne dass mehr Speicher übrig war.

Die Lösung bestand darin, entweder Swap hinzuzufügen oder den Worker / Executor so zu konfigurieren, dass zusätzlich zur Verwendung der Speicherebene MEMORY_AND_DISK für mehrere Persists weniger Speicher verwendet wird.

Joren Van Severen
quelle
3
Wenn Sie eine Ressource auf dem Knoten (Speicher) haben, können Sie versuchen, den Spark Executor-Speicher zu erhöhen. Ich werde das zuerst versuchen, wenn Sie auch über die Leistung besorgt sind.
Nir
14
Hi @Joren das ist kein Wettbewerb. Das OP-Problem besteht darin, dass der Executor nicht über genügend Speicher verfügt, um die Shuffle-Ausgabe zu speichern. Was für Sie funktioniert hat, ist nicht, den Executor-Speicher zu verringern, sondern die Speicherebene MEMORY_AND_DISK zu verwenden, wodurch die Speicherbeschränkung des Executors beseitigt wird. Auch OP sagt nicht aus, wie viel Ressource er für Executor hat.
Nir
Ich habe das gleiche Problem und habe Methoden wie das Erhöhen des Executor-Speichers, das Erhöhen der Anzahl der Partitionen und das Freigeben von mehr physischem Speicher ausprobiert. Und manchmal hat es funktioniert, manchmal nicht. Ich habe festgestellt, dass dies nur in der Shuffle-Lesephase passiert ist, und möchte fragen, wo ich den StorageLevel einstellen kann.
Lhfcws
Ich habe meine Datenstruktur optimiert und korrigiert. Ich habe gerade HashMap in ein Byte [] geändert, das von Protostuff serialisiert wurde
Lhfcws
1
Versuchen Sie, spark.driver.overhead.memory und spark.executor.overhead.memory auf einen Wert von mehr als 384 (Standard) zu ändern, und es sollte funktionieren. Sie können entweder 1024 MB oder 2048 MB verwenden.
Rahul Gulati
14

Wir hatten einen ähnlichen Fehler mit Spark, aber ich bin nicht sicher, ob er mit Ihrem Problem zusammenhängt.

Wir haben JavaPairRDD.repartitionAndSortWithinPartitions100 GB Daten verwendet und es ist ähnlich wie bei Ihrer App immer wieder fehlgeschlagen. Dann haben wir uns die Garnprotokolle auf den spezifischen Knoten angesehen und festgestellt, dass wir ein Problem mit nicht genügend Speicher haben, sodass das Garn die Ausführung unterbrochen hat. Unsere Lösung war zu ändern / spark.shuffle.memoryFraction 0in .../spark/conf/spark-defaults.conf. Dadurch konnten wir auf diese Weise eine viel größere (aber leider nicht unendliche) Datenmenge verarbeiten.

Notinlist
quelle
Ist es wirklich "0" oder war das ein Tippfehler? Welche Logik steckt dahinter, um zu erzwingen, dass es dauerhaft auf die Festplatte gelangt?
Virgil
@ Virgil Ja. Wir haben einige Tests gemacht. Je näher wir Null waren, desto größer wurde die verarbeitbare Menge. Der Preis betrug 20% ​​der Zeit.
Notinlist
Interessanterweise reduziere ich auch spark.shuffle.memoryFraction auf Null, habe aber mehr Fehler hintereinander. (Nämlich: MetadataFetchFailedException und FetchFailedException intermittierend) Es sollte ein Fehler / Problem werden, wenn "All-Spill" weniger Fehler aufweist als "Partial-Spill".
Tribbloid
10

Ich habe das gleiche Problem auf meinem 3-Computer-YARN-Cluster. Ich habe den Arbeitsspeicher ständig gewechselt, aber das Problem blieb bestehen. Schließlich sah ich die folgenden Meldungen in den Protokollen:

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

und danach gab es diese Nachricht:

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

Ich habe die Eigenschaften in spark-defaults.conf wie folgt geändert:

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

Das ist es! Mein Job wurde danach erfolgreich abgeschlossen.

xplorerdev
quelle
In den Funken-Dokumenten heißt es : spark.executor.heartbeatInterval should be significantly less than spark.network.timeout. Daher ist es möglicherweise nicht die beste Idee, beide auf denselben Wert zu setzen.
Bitswazsky
2

Für mich machte ich ein Fenster mit großen Datenmengen (ungefähr 50B Zeilen) und bekam eine Bootsladung von

ExternalAppendOnlyUnsafeRowArray:54 - Überlaufschwelle von 4096 Zeilen erreicht, Umschalten auf org.apache.spark.util.collection.unsafe.sort.UnsafeExternalSorter

In meinen Protokollen. Offensichtlich kann 4096 bei einer solchen Datengröße klein sein ... dies führte mich zu folgendem JIRA:

https://issues.apache.org/jira/browse/SPARK-21595

Und letztendlich zu den folgenden zwei Konfigurationsoptionen:

  • spark.sql.windowExec.buffer.spill.threshold
  • spark.sql.windowExec.buffer.in.memory.threshold

Beide sind standardmäßig 4096; Ich habe sie viel höher angehoben (2097152) und die Dinge scheinen jetzt gut zu laufen. Ich bin mir nicht 100% sicher, ob dies das gleiche ist wie das hier angesprochene Problem, aber es ist eine andere Sache, die Sie versuchen sollten.

MichaelChirico
quelle
1

Ich habe diesen Fehler behoben und den zugewiesenen Speicher in executorMemory und driverMemory erhöht. Sie können dies in HUE tun, indem Sie das Spark-Programm auswählen, das das Problem verursacht, und in den Eigenschaften -> Optionsliste können Sie Folgendes hinzufügen:

--driver-memory 10G --executor-memory 10G --num-executors 50 --executor-cores 2

Natürlich variieren die Werte der Parameter abhängig von der Größe Ihres Clusters und Ihren Anforderungen.

Ignacio Alorre
quelle
1

Wenn in der Spark-Web-Benutzeroberfläche Informationen wie vorhanden sind Executors lost, müssen Sie das Garnprotokoll überprüfen und sicherstellen, dass Ihr Container getötet wurde.

Wenn der Container getötet wurde, liegt dies wahrscheinlich an dem Mangel an Speicher.

Wie finde ich die wichtigsten Informationen in Garnprotokollen? Zum Beispiel könnte es einige Warnungen wie diese geben:

Container killed by YARN for exceeding memory limits. 2.5 GB of 2.5 GB physical memory used. 
Consider boosting spark.yarn.executor.memoryOverhead.

In diesem Fall sollten Sie erhöhen spark.yarn.executor.memoryOverhead.

DennisLi
quelle
0

In meinem Fall (Standalone-Cluster) wurde die Ausnahme ausgelöst, da das Dateisystem einiger Spark-Slaves zu 100% gefüllt war. Das Löschen aller Elemente in den spark/workOrdnern der Slaves löste das Problem.

i000174
quelle
0

Ich habe das gleiche Problem, aber ich habe viele Antworten gesucht, die mein Problem nicht lösen können. Schließlich debugge ich meinen Code Schritt für Schritt. Ich finde, dass das Problem, das durch die Datengröße verursacht wird, nicht für jede Partition ausgeglichen ist, was dazu führt, MetadataFetchFailedExceptiondass in mapPhase nicht reducePhase. einfach df_rdd.repartition(nums)vorher machenreduceByKey()

Hund
quelle