FetchFailedException veya MetadataFetchFailedException

Question

Ayrıştırma kodunu 1 GB veri kümesiyle çalıştırdığımda herhangi bir hata olmadan tamamlanır. Ancak, bir seferde 25 gb veriyi denediğimde hataların altına iniyorum. Başarısızlıkların altında nasıl kaçınıyorum anlamaya çalışıyorum. Herhangi bir öneri veya fikir duymaktan mutluluk duyarız.

Differnt hataları

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

org.apache.spark.shuffle.FetchFailedException: Failed to connect to ip-xxxxxxxx 

org.apache.spark.shuffle.FetchFailedException: Error in opening FileSegmentManagedBuffer{file=/mnt/yarn/nm/usercache/xxxx/appcache/application_1450751731124_8446/blockmgr-8a7b17b8-f4c3-45e7-aea8-8b0a7481be55/08/shuffle_0_224_0.data, offset=12329181, length=2104094} 

Küme Detayları:

İplik: 8 Düğümler
Toplam çekirdekler: 64
Bellek: 500 GB
Kıvılcım Versiyon: 1,5

Spark gönderi beyanı : Yığın izlemesi

spark-submit --master yarn-cluster \ 
         --conf spark.dynamicAllocation.enabled=true \ 
         --conf spark.shuffle.service.enabled=true \ 
         --executor-memory 4g \ 
         --driver-memory 16g \ 
         --num-executors 50 \ 
         --deploy-mode cluster \ 
         --executor-cores 1 \ 
         --class my.parser \ 
         myparser.jar \ 
         -input xxx \ 
         -output xxxx \ 

Bir:

at org.apache.spark.MapOutputTracker$$anonfun$org$apache$spark$MapOutputTracker$$convertMapStatuses$2.apply(MapOutputTracker.scala:460) 
at org.apache.spark.MapOutputTracker$$anonfun$org$apache$spark$MapOutputTracker$$convertMapStatuses$2.apply(MapOutputTracker.scala:456) 
at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772) 
at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33) 
at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:108) 
at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771) 
at org.apache.spark.MapOutputTracker$.org$apache$spark$MapOutputTracker$$convertMapStatuses(MapOutputTracker.scala:456) 
at org.apache.spark.MapOutputTracker.getMapSizesByExecutorId(MapOutputTracker.scala:183) 
at org.apache.spark.shuffle.hash.HashShuffleReader.read(HashShuffleReader.scala:47) 
at org.apache.spark.rdd.ShuffledRDD.compute(ShuffledRDD.scala:90) 
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297) 
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264) 
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) 
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:297) 
at org.apache.spark.rdd.RDD.iterator(RDD.scala:264) 
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66) 
at org.apache.spark.scheduler.Task.run(Task.scala:88) 
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214) 
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 
at java.lang.Thread.run(Thread.java:745) 

Answer 1

Bu hata neredeyse uygulayıcıları bellek sorunları nedeniyle garanti edilmektedir. Bu tür problemleri çözmek için birkaç yol düşünebilirim.

1) Daha fazla bölümle çalışmayı deneyebilirsiniz (, dataframe ürününüzde). Bellek sorunları genellikle bir veya daha fazla bölüm belleğe sığacak daha fazla veri içerdiğinde ortaya çıkar.

2) Açık bir şekilde spark.yarn.executor.memoryOverhead ayarlamamış olduğunuzu fark ettim, bu nedenle durumda 400MB olacak max(386, 0.10* executorMemory) için varsayılan olacaktır. Bu benim için düşük geliyor. 1GB diyelim ki, bunu 1GB'a kadar artırmaya çalışacağım (eğer bellek 1GB'a yükselirseniz, --executor-memory'u 3GB'a düşürmeniz gerekir)

3) Arızalı düğümlerdeki günlük dosyalarına bakın. "Öldürme kabı" metnini aramak istersiniz. "Fiziksel hafıza sınırlarının ötesine geçen" metnini görürseniz, memoryOverhead'i artırmak - benim tecrübemde - sorunu çözer.

Answer 2

Ayrıca, spark.network.timeout Spark zaman aşımını 800 gibi daha büyük bir değere yükselterek bazı iyi sonuçlar aldım. Varsayılan 120 saniye, çok fazla sayıda yürütücünün ağır yük altındayken zaman aşımına uğramasına neden olacaktır.