TimeoutException almak için olası nedenleri nelerdir: Vadeli sonra zaman aşımına [n saniye] Spark

Question

Bir Kıvılcım SQL programı üzerinde çalışıyorum ve aşağıdaki özel durum alıyorum çalışırken:

16/11/07 15:58:25 ERROR yarn.ApplicationMaster: User class threw exception: java.util.concurrent.TimeoutException: Futures timed out after [3000 seconds] 
java.util.concurrent.TimeoutException: Futures timed out after [3000 seconds] 
    at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:219) 
    at scala.concurrent.impl.Promise$DefaultPromise.result(Promise.scala:223) 
    at scala.concurrent.Await$$anonfun$result$1.apply(package.scala:190) 
    at scala.concurrent.BlockContext$DefaultBlockContext$.blockOn(BlockContext.scala:53) 
    at scala.concurrent.Await$.result(package.scala:190) 
    at org.apache.spark.sql.execution.joins.BroadcastHashJoin.doExecute(BroadcastHashJoin.scala:107) 
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:132) 
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:130) 
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150) 
    at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:130) 
    at org.apache.spark.sql.execution.Project.doExecute(basicOperators.scala:46) 
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:132) 
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:130) 
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150) 
    at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:130) 
    at org.apache.spark.sql.execution.Union$$anonfun$doExecute$1.apply(basicOperators.scala:144) 
    at org.apache.spark.sql.execution.Union$$anonfun$doExecute$1.apply(basicOperators.scala:144) 
    at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245) 
    at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245) 
    at scala.collection.immutable.List.foreach(List.scala:381) 
    at scala.collection.TraversableLike$class.map(TraversableLike.scala:245) 
    at scala.collection.immutable.List.map(List.scala:285) 
    at org.apache.spark.sql.execution.Union.doExecute(basicOperators.scala:144) 
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:132) 
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:130) 
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150) 
    at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:130) 
    at org.apache.spark.sql.execution.columnar.InMemoryRelation.buildBuffers(InMemoryColumnarTableScan.scala:129) 
    at org.apache.spark.sql.execution.columnar.InMemoryRelation.<init>(InMemoryColumnarTableScan.scala:118) 
    at org.apache.spark.sql.execution.columnar.InMemoryRelation$.apply(InMemoryColumnarTableScan.scala:41) 
    at org.apache.spark.sql.execution.CacheManager$$anonfun$cacheQuery$1.apply(CacheManager.scala:93) 
    at org.apache.spark.sql.execution.CacheManager.writeLock(CacheManager.scala:60) 
    at org.apache.spark.sql.execution.CacheManager.cacheQuery(CacheManager.scala:84) 
    at org.apache.spark.sql.DataFrame.persist(DataFrame.scala:1581) 
    at org.apache.spark.sql.DataFrame.cache(DataFrame.scala:1590) 
    at com.somecompany.ml.modeling.NewModel.getTrainingSet(FlowForNewModel.scala:56) 
    at com.somecompany.ml.modeling.NewModel.generateArtifacts(FlowForNewModel.scala:32) 
    at com.somecompany.ml.modeling.Flow$class.run(Flow.scala:52) 
    at com.somecompany.ml.modeling.lowForNewModel.run(FlowForNewModel.scala:15) 
    at com.somecompany.ml.Main$$anonfun$2.apply(Main.scala:54) 
    at com.somecompany.ml.Main$$anonfun$2.apply(Main.scala:54) 
    at scala.Option.getOrElse(Option.scala:121) 
    at com.somecompany.ml.Main$.main(Main.scala:46) 
    at com.somecompany.ml.Main.main(Main.scala) 
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 
    at java.lang.reflect.Method.invoke(Method.java:498) 
    at org.apache.spark.deploy.yarn.ApplicationMaster$$anon$2.run(ApplicationMaster.scala:542) 
16/11/07 15:58:25 INFO yarn.ApplicationMaster: Final app status: FAILED, exitCode: 15, (reason: User class threw exception: java.util.concurrent.TimeoutException: Futures timed out after [3000 seconds]) 

son Kod izimin yığın izinden tanıdığım parçanın bir kısmı com.somecompany.ml.modeling.NewModel.getTrainingSet(FlowForNewModel.scala:56) bu satırda beni alır: profilesDF.cache() Önbelleğe alma işleminden önce 2 veri karesi arasında bir birleşimi yapıyorum. Ben here katılmadan önce hem de veri kareleri ısrar hakkında bir cevap gördüm Ben hala benim dönüşümler

kullanıyorum beri sendikalanan veri alanını önbelleğe ihtiyacım var Ve bu istisna neden olabilir merak ediyordum ? Bunun için arama, bana bir bağlantı için rpc zaman aşımı istisnası ya da benim sorunum değil bazı güvenlik sorunları ile ilgili var Eğer nasıl çözüleceğine dair herhangi bir fikriniz varsa Açıkçası bunu takdir ediyorum ama sadece sorunu anlamak bana yardımcı olacaktır önceden

Teşekkür

Answer 1

Soru çözmek: Ben atılmasına bu duruma neden olabilir merak ediyorum? Tüm ağ etkileşimleri için

spark.sql.broadcastTimeout 300 Timeout in seconds for the broadcast wait time in broadcast joins

spark.network.timeout 120s Standart zaman aşımı .. spark.network.timeout (spark.rpc.askTimeout), spark.sql.broadcastTimeout, spark.kryoserializer.buffer.max (eğer Kryo seri kullanıyorsanız), vb ile akort edilir: Cevap

Karmaşık sorguları işlemek için 'daki büyük varsayılan değerler. Bu değerlerle başlayabilir ve SQL iş yüklerinize göre ayarını yapabilirsiniz.

Not: Doc says that

aşağıdaki seçenekler (. Spark.sql bakınız özellikler) da ayarlamak için sorgu yürütme performansını kullanılabilir. Daha fazla optimizasyon otomatik olarak gerçekleştirilirken, bu seçeneklerin daha sonraki sürümlerde kullanımdan kaldırılması mümkündür. *

Ayrıca, daha iyi anlaşmanız için, yukarıdaki yığın izlemesi için yürütme yönteminin tetikleyici noktası olan BroadCastHashJoin'u görebilirsiniz.

protected override def doExecute(): RDD[Row] = { 
    val broadcastRelation = Await.result(broadcastFuture, timeout) 

    streamedPlan.execute().mapPartitions { streamedIter => 
     hashJoin(streamedIter, broadcastRelation.value) 
    } 
    } 

Answer 2

Ram'ın önerisinin bazı durumlarda çalıştığını bilmek güzel. Bu istisna üzerine birkaç kez tökezlediğimi belirtmek isterim (here tarif edilen dahil).

Çoğu zaman, bazı yürütücülerdeki neredeyse sessiz olan OOM'lardan kaynaklanıyordu. Başarısız görevler için SparkUI'ye bakın, bu tablonun son sütunu: OOM mesajlarını görebilirsiniz.

İyi bir kıvılcımla iç içe olanları anlarsanız, yayınlanan veriler sürücüden geçer. Bu nedenle, sürücü, vericilerin verilerini toplamak ve onu herkese geri göndermek için bazı mekanizmalara sahiptir. Bir noktada bir yönetici başarısız olursa, bu zaman aşımları ile sonuçlanabilir.