优雅关闭

  1. package com.atguigu.bigdata.spark.streaming
  3. import org.apache.spark.SparkConf
  4. import org.apache.spark.streaming.dstream.DStream
  5. import org.apache.spark.streaming.{Seconds, StreamingContext, StreamingContextState}
  7. object SparkStreaming08_Close {
  9.     def main(args: Array[String]): Unit = {
  11.         /*
  12.            线程的关闭：
  13.            val thread = new Thread()
  14.            thread.start()
  15.            thread.stop(); // 强制关闭
  16.          */
  17.         val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
  18.         val ssc = new StreamingContext(sparkConf, Seconds(3))
  20.         val lines = ssc.socketTextStream("localhost", 9999)
  21.         val wordToOne = lines.map((_,1))
  23.         wordToOne.print()
  25.         ssc.start()
  27.         // 如果想要关闭采集器，那么需要创建新的线程
  28.         // 而且需要在第三方程序中增加关闭状态
  29.         new Thread(
  30.             new Runnable {
  31.                 override def run(): Unit = {
  32.                     // 优雅地关闭
  33.                     // 计算节点不在接收新的数据，而是将现有的数据处理完毕，然后关闭
  34.                     // Mysql : Table(stopSpark) => Row => data
  35.                     // Redis : Data（K-V）
  36.                     // ZK    : /stopSpark
  37.                     // HDFS  : /stopSpark
  38.                     /*
  39.                     while ( true ) {
  40.                         if (true) {
  41.                             // 获取SparkStreaming状态
  42.                             val state: StreamingContextState = ssc.getState()
  43.                             if ( state == StreamingContextState.ACTIVE ) {
  44.                                 ssc.stop(true, true)
  45.                             }
  46.                         }
  47.                         Thread.sleep(5000)
  48.                     }
  49.                      */
  51.                     Thread.sleep(5000)
  52.                     val state: StreamingContextState = ssc.getState()
  53.                     if ( state == StreamingContextState.ACTIVE ) {
  54.                         ssc.stop(true, true)
  55.                     }
  56.                     System.exit(0)
  57.                 }
  58.             }
  59.         ).start()
  61.         ssc.awaitTermination() // block 阻塞main线程
  63.     }
  64. }

优雅恢复

  1. package com.atguigu.bigdata.spark.streaming
  3. import org.apache.spark.SparkConf
  4. import org.apache.spark.streaming.{Seconds, StreamingContext, StreamingContextState}
  6. object SparkStreaming09_Resume {
  8.     def main(args: Array[String]): Unit = {
  10.         val ssc = StreamingContext.getActiveOrCreate("cp", ()=>{
  11.             val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
  12.             val ssc = new StreamingContext(sparkConf, Seconds(3))
  14.             val lines = ssc.socketTextStream("localhost", 9999)
  15.             val wordToOne = lines.map((_,1))
  17.             wordToOne.print()
  19.             ssc
  20.         })
  22.         ssc.checkpoint("cp")
  24.         ssc.start()
  25.         ssc.awaitTermination() // block 阻塞main线程
  26.     }
  27. }