在大型數據集上運行spark時，「sparkContext已關閉」

在羣集上運行sparkJob超過特定數據大小（〜2,5gb）時，我得到「因爲關閉了SparkContext而導致作業被取消」或「執行程序丟失」。看着紗I，我看到那個被殺的工作是成功的。運行500MB數據時沒有問題。我正在尋找解決方案，發現： - 「似乎紗線殺死了一些執行者，因爲他們要求的內存超出預期。」在大型數據集上運行spark時，「sparkContext已關閉」

如何調試它的任何建議？

命令我提交我的火花的工作有：

/opt/spark-1.5.0-bin-hadoop2.4/bin/spark-submit --driver-memory 22g --driver-cores 4 --num-executors 15 --executor-memory 6g --executor-cores 6 --class sparkTesting.Runner --master yarn-client myJar.jar jarArguments

和sparkContext設置

val sparkConf = (new SparkConf() 
    .set("spark.driver.maxResultSize", "21g") 
    .set("spark.akka.frameSize", "2011") 
    .set("spark.eventLog.enabled", "true") 
    .set("spark.eventLog.enabled", "true") 
    .set("spark.eventLog.dir", configVar.sparkLogDir) 
    )

簡化代碼失敗看起來像

val hc = new org.apache.spark.sql.hive.HiveContext(sc) 
val broadcastParser = sc.broadcast(new Parser()) 

val featuresRdd = hc.sql("select "+ configVar.columnName + " from " + configVar.Table +" ORDER BY RAND() LIMIT " + configVar.Articles) 
val myRdd : org.apache.spark.rdd.RDD[String] = featuresRdd.map(doSomething(_,broadcastParser)) 

val allWords= featuresRdd 
    .flatMap(line => line.split(" ")) 
    .count 

val wordQuantiles= featuresRdd 
    .flatMap(line => line.split(" ")) 
    .map(word => (word, 1)) 
    .reduceByKey(_ + _) 
    .map(pair => (pair._2 , pair._2)) 
    .reduceByKey(_+_) 
    .sortBy(_._1) 
    .collect 
    .scanLeft((0,0.0)) ((res,add) => (add._1, res._2+add._2)) 
    .map(entry => (entry._1,entry._2/allWords)) 

val dictionary = featuresRdd 
    .flatMap(line => line.split(" ")) 
    .map(word => (word, 1)) 
    .reduceByKey(_ + _) // here I have Rdd of word,count tuples 
    .filter(_._2 >= moreThan) 
    .filter(_._2 <= lessThan) 
    .filter(_._1.trim!=("")) 
    .map(_._1) 
    .zipWithIndex 
    .collect 
    .toMap

和錯誤堆棧

Exception in thread "main" org.apache.spark.SparkException: Job cancelled because SparkContext was shut down 
at org.apache.spark.scheduler.DAGScheduler$$anonfun$cleanUpAfterSchedulerStop$1.apply(DAGScheduler.scala:703) 
at org.apache.spark.scheduler.DAGScheduler$$anonfun$cleanUpAfterSchedulerStop$1.apply(DAGScheduler.scala:702) 
at scala.collection.mutable.HashSet.foreach(HashSet.scala:79) 
at org.apache.spark.scheduler.DAGScheduler.cleanUpAfterSchedulerStop(DAGScheduler.scala:702) 
at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onStop(DAGScheduler.scala:1511) 
at org.apache.spark.util.EventLoop.stop(EventLoop.scala:84) 
at org.apache.spark.scheduler.DAGScheduler.stop(DAGScheduler.scala:1435) 
at org.apache.spark.SparkContext$$anonfun$stop$7.apply$mcV$sp(SparkContext.scala:1715) 
at org.apache.spark.util.Utils$.tryLogNonFatalError(Utils.scala:1185) 
at org.apache.spark.SparkContext.stop(SparkContext.scala:1714) 
at org.apache.spark.scheduler.cluster.YarnClientSchedulerBackend$MonitorThread.run(YarnClientSchedulerBackend.scala:146) 
at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:567) 
at org.apache.spark.SparkContext.runJob(SparkContext.scala:1813) 
at org.apache.spark.SparkContext.runJob(SparkContext.scala:1826) 
at org.apache.spark.SparkContext.runJob(SparkContext.scala:1839) 
at org.apache.spark.SparkContext.runJob(SparkContext.scala:1910) 
at org.apache.spark.rdd.RDD.count(RDD.scala:1121) 
at sparkTesting.InputGenerationAndDictionaryComputations$.createDictionary(InputGenerationAndDictionaryComputations.scala:50) 
at sparkTesting.Runner$.main(Runner.scala:133) 
at sparkTesting.Runner.main(Runner.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:483) 
at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:672) 
at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:180) 
at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:205) 
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:120) 
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

來源

2015-09-28 Aleksander Zendel

根據我的經驗，這幾乎總是由於OOM例外。嘗試查看各個執行器機器上的日誌文件。 –

我會從您的作業中打印堆棧並使用一些Java util工具監視JVM堆大小：jstat，jstatd，jconsole ...以瞭解有關此限制的更多信息。如果你還有物理內存，你可以在啓動你的應用之前增加JVM的內存大小！您可以根據您的優化堆大小調整您的集合。 –