大きなデータセットを処理するときにFetchFailedExceptionまたはMetadataFetchFailedExceptionが発生する

Question

1GBデータセットで解析コードを実行すると、エラーは発生しません。しかし、一度に25GBのデータを試してみると、私はエラー以下になります。私はどのようにして失敗を避けることができるのか理解しようとしています。何か提案やアイデアを聞いてうれしいです。

Differntエラー、

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} 

クラスタの詳細：

糸：8つのノード
総コア：64
メモリ：500ギガバイト
スパークバージョン：1.5

Spark submit statement ：スタックトレースの

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 \ 

ワン：

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

このエラーは、ほとんど自分の執行上のメモリの問題によって引き起こされることが保証されます。私はこれらのタイプの問題に対処する2つの方法を考えることができます。

1）多くのパーティションで実行できます（dataframeにrepartitionを実行してください）。メモリの問題は、通常、1つ以上のパーティションにメモリよりも多くのデータが含まれている場合に発生します。

2）あなたが明示的にspark.yarn.executor.memoryOverheadを設定していないことに気付いていますので、デフォルトでmax(386, 0.10* executorMemory)になります。この場合は400MBになります。それは私には聞こえません。私は1GBと言うようにそれを増やそうとしています。（メモリオーバーヘッドを1GBに増やすと、--executor-memoryを3GBに下げる必要があります）3）障害のあるノードのログファイルを調べます。 「殺すコンテナ」のテキストを探したいとします。 「物理メモリの限界を超えて実行しています」というテキストが表示された場合、私の経験上、memoryOverheadを増やすことで問題が解決されます。

Answer 2

また、スパークタイムアウトspark.network.timeoutを800などの大きな値に増やしても良い結果が得られました。デフォルトの120秒は、重い負荷がかかっているときに多くのエグゼキュータがタイムアウトする原因になります。