优雅关闭SparkStreaming程序

素颜马尾好姑娘i 2022-10-24 13:28 227阅读 0赞

# 优雅关闭SparkStreaming的程序 #

流式任务需要7\*24小时执行，但是有时涉及到升级代码需要主动停止程序，但是分布式程序，没办法做到一个个进程去杀死，所有配置优雅的关闭就显得至关重要了。  
使用外部文件系统来控制内部程序关闭。

大致思路如下：**新启动一个线程检测hdfs上是否多出了一个目录，检测到后使用优雅的关闭方式，先停止采集数据，然后将已经采集到的数据计算完毕，然后关闭一切连接。**

@Test
      def getclose():Unit={ 
      
        streamingContext.start()
        
        streamingContext.stop()
      
        new Thread(){ 
          //新开一个线程，判断何时需要关闭
          override def run(): Unit = { 
            //开启守护线程
            setDaemon(true)
            
            //定义函数，获取是否需要关闭线程
            def getNoCloseSingal():Boolean={ 
              //判断是否需要关闭线程的逻辑
              true
              false
            }
            
            //不断查询是否需要关闭
            while (getNoCloseSingal){ 
              
              //停顿一秒，继续查询是否需要关闭
              Thread.sleep(1000)
              
            }
            
            //执行关闭操作
            streamingContext.stop(true,true)
            
          }
        }.start()
        
      }

*  MonitorStop

import java.net.URI
    
    import org.apache.hadoop.conf.Configuration
    import org.apache.hadoop.fs.{ FileSystem, Path}
    import org.apache.spark.streaming.{ StreamingContext, StreamingContextState}
    
    class MonitorStop(ssc: StreamingContext) extends Runnable { 
    
      override def run(): Unit = { 
    
        val fs: FileSystem = FileSystem.get(new URI("hdfs://linux1:9000"), new Configuration(), "atguigu")
    
        while (true) { 
          try
            Thread.sleep(5000)
          catch { 
            case e: InterruptedException =>
              e.printStackTrace()
          }
          val state: StreamingContextState = ssc.getState
    
          val bool: Boolean = fs.exists(new Path("hdfs://linux1:9000/stopSpark"))
    
          if (bool) { 
            if (state == StreamingContextState.ACTIVE) { 
              ssc.stop(stopSparkContext = true, stopGracefully = true)
              System.exit(0)
            }
          }
        }
      }
    }

*  SparkTest

import org.apache.spark.SparkConf
    import org.apache.spark.streaming.dstream.{ DStream, ReceiverInputDStream}
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    object SparkTest { 
    
      def createSSC(): _root_.org.apache.spark.streaming.StreamingContext = { 
    
        val update: (Seq[Int], Option[Int]) => Some[Int] = (values: Seq[Int], status: Option[Int]) => { 
    
          //当前批次内容的计算
          val sum: Int = values.sum
    
          //取出状态信息中上一次状态
          val lastStatu: Int = status.getOrElse(0)
    
          Some(sum + lastStatu)
        }
    
        val sparkConf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("SparkTest")
    
        //设置优雅的关闭
        sparkConf.set("spark.streaming.stopGracefullyOnShutdown", "true")
    
        val ssc = new StreamingContext(sparkConf, Seconds(5))
    
        ssc.checkpoint("./ck")
    
        val line: ReceiverInputDStream[String] = ssc.socketTextStream("linux1", 9999)
    
        val word: DStream[String] = line.flatMap(_.split(" "))
    
        val wordAndOne: DStream[(String, Int)] = word.map((_, 1))
    
        val wordAndCount: DStream[(String, Int)] = wordAndOne.updateStateByKey(update)
    
        wordAndCount.print()
    
        ssc
      }
    
      def main(args: Array[String]): Unit = { 
    
        val ssc: StreamingContext = StreamingContext.getActiveOrCreate("./ck", () => createSSC())
    
        new Thread(new MonitorStop(ssc)).start()
    
        ssc.start()
        ssc.awaitTermination()
      }
    }