spark大数据分析:sparkStrreaming(21)程序关闭

文章目录关闭程序方法有4种1.kill杀死,可能数据丢失2,通过钩子(hook)关闭,需要写shell脚本,麻烦3.在程序中建立http服务,接受外部消息在程序中关闭,代码较多4.用hdfs中目录做标记,定期检查hdfs目录是否存在,存在关闭程序,简单方便依赖<dependency><groupId>org.apache.hadoop</groupId><

_东极

424人浏览 · 2020-09-02 00:19:03

_东极 · 2020-09-02 00:19:03 发布

文章目录

关闭程序方法有4种
1.kill杀死,可能数据丢失
2,通过钩子(hook)关闭,需要写shell脚本,麻烦
3.在程序中建立http服务,接受外部消息在程序中关闭,代码较多
4.用hdfs中目录做标记,定期检查hdfs目录是否存在,存在关闭程序,简单方便

依赖

   <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>2.7.4</version>
        </dependency>

package stream

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.FileSystem
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.reflect.io.Path

object Test08 {
  var hadoopConf: Configuration = _
  val shutdownMarkerPath = new Path("hdfs://linux01:8020/user/admin/tmp/spark_shutdown_marker")
  var stopMarker: Boolean = false
  def main(args: Array[String]) {
    val conf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("Chapter8_10_2")
    //      .set("spark.streaming.stopGracefullyOnShutdown", "true")
    val ssc = new StreamingContext(conf, Seconds(5))
    ssc.sparkContext.setLogLevel("WARN")

    hadoopConf = ssc.sparkContext.hadoopConfiguration
    hadoopConf.set("fs.defaultFS", "hdfs://linux01:8020")

    val lines = ssc.socketTextStream("linux01", 9999)
    val words = lines.flatMap(_.split(" "))
    val wordMap = words.map(x => (x, 1))
    val wordCounts = wordMap.reduceByKey(_ + _)
    wordCounts.print()

    ssc.start()
    val checkIntervalMillis = 2 * 1000 * 1
    var isStopped = false

    while (!isStopped) {
      println("正在确认关闭状态: ")
      isStopped = ssc.awaitTerminationOrTimeout(checkIntervalMillis)
      if (isStopped)
        println("Spark Streaming Chapter8_10已关闭.")
      else
        println("Spark Streaming Chapter8_10运行中...")
      checkShutdownMarker
      if (!isStopped && stopMarker) {
        println("准备关闭Spark Streaming")
        ssc.stop(true, true)
      }
    }
  }

  def checkShutdownMarker = {
    if (!stopMarker) {
      val fs = FileSystem.get(hadoopConf)
      stopMarker = fs.exists(shutdownMarkerPath)
    }
  }
}