[jira] [Comment Edited] (SPARK-19264) Work should start driver, the same to AM of yarn

["hustfxj (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/SPARK-19264?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15827181#comment-15827181 ] 

hustfxj edited comment on SPARK-19264 at 1/18/17 1:12 AM:
----------------------------------------------------------

[~srowen] sorry, I didn't say it clearly. I means an spark application can't be done when it contains other unfinished non-daemon. for example
{code:title=Test.scala|borderStyle=solid}   
   val sparkConf = new SparkConf().setAppName("NetworkWordCount")
    sparkConf.set("spark.streaming.blockInterval", "1000ms")
    val ssc = new StreamingContext(sparkConf, Seconds(10))
    //non-daemon thread
    val scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(
      new ThreadFactory() {
        def newThread(r: Runnable): Thread = new Thread(r, "Driver-Commit-Thread")
      })
    scheduledExecutorService.scheduleAtFixedRate(
      new Runnable() {
        def run() {
          try {
            System.out.println("runable")
          } catch {
            case e: Exception => {
              System.out.println("ScheduledTask persistAllConsumerOffset exception", e)
            }
          }
        }
      }, 1000, 1000 * 5, TimeUnit.MILLISECONDS)
    val lines = ssc.receiverStream(new WordReceiver(StorageLevel.MEMORY_AND_DISK_2))
    val words = lines.flatMap(_.split(" "))
    val wordCounts = words.map(x => (x, 1)).reduceByKey((x: Int, y: Int) => x + y, 10)
    wordCounts.foreachRDD{rdd =>
      rdd.collect().foreach(println)
      throw new RuntimeException
    }
    ssc.start()
    ssc.awaitTermination()
{code}




was (Author: hustfxj):
[~srowen] sorry, I didn't say it clearly. I means an spark application can't be done when it contain other unfinished non-daemon. for example
{code:title=Test.scala|borderStyle=solid}   
   val sparkConf = new SparkConf().setAppName("NetworkWordCount")
    sparkConf.set("spark.streaming.blockInterval", "1000ms")
    val ssc = new StreamingContext(sparkConf, Seconds(10))
    //non-daemon thread
    val scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(
      new ThreadFactory() {
        def newThread(r: Runnable): Thread = new Thread(r, "Driver-Commit-Thread")
      })
    scheduledExecutorService.scheduleAtFixedRate(
      new Runnable() {
        def run() {
          try {
            System.out.println("runable")
          } catch {
            case e: Exception => {
              System.out.println("ScheduledTask persistAllConsumerOffset exception", e)
            }
          }
        }
      }, 1000, 1000 * 5, TimeUnit.MILLISECONDS)
    val lines = ssc.receiverStream(new WordReceiver(StorageLevel.MEMORY_AND_DISK_2))
    val words = lines.flatMap(_.split(" "))
    val wordCounts = words.map(x => (x, 1)).reduceByKey((x: Int, y: Int) => x + y, 10)
    wordCounts.foreachRDD{rdd =>
      rdd.collect().foreach(println)
      throw new RuntimeException
    }
    ssc.start()
    ssc.awaitTermination()
{code}



> Work should start driver, the same to  AM  of yarn 
> ---------------------------------------------------
>
>                 Key: SPARK-19264
>                 URL: https://issues.apache.org/jira/browse/SPARK-19264
>             Project: Spark
>          Issue Type: Improvement
>            Reporter: hustfxj
>
>   I think work can't start driver by "ProcessBuilderLike",  thus we can't know the application's main thread is finished or not if the application's main thread contains some non-daemon threads. Because the program terminates when there no longer is any non-daemon thread running (or someone called System.exit). The main thread can have finished long ago. 
>     worker should  start driver like AM of YARN . As followed:
> {code:title=ApplicationMaster.scala|borderStyle=solid}    
>      mainMethod.invoke(null, userArgs.toArray)
>      finish(FinalApplicationStatus.SUCCEEDED, ApplicationMaster.EXIT_SUCCESS)
>      logDebug("Done running users class")
> {code}
> Then the work can monitor the driver's main thread, and know the application's state. 



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org