import org.apache.http.client.methods.HttpGet
import org.apache.http.impl.client.DefaultHttpClient
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Duration, StreamingContext}

/**
  * 注册SparkStreaming
  */
object StreamingMonitor{

  /**
    * 创建streamingContext
    *
    * @param args     参数
    * @param duration 窗口
    * @param name     名字
    * @return
    */
  def createContext(args:Array[String],duration:Duration,name:String):StreamingContext = {
    val sparkConf = new SparkConf().setAppName(name)
    createContext(args,sparkConf,duration)
  }

  /**
    * 创建streamingContext
    *
    * @param args     参数
    * @param conf     配置
    * @param duration 窗口
    * @return
    */
  def createContext(args:Array[String], conf: SparkConf, duration:Duration):StreamingContext = {
    println("*************** register begin **************")

    println("args:")
    args.foreach(println)
    println()

    System.setProperty("hive.metastore.uris", "thrift://hnode1:9083")

    val ssc = new StreamingContext(conf,duration)
    ssc.sparkContext.setLogLevel("warn")

    val appId = ssc.sparkContext.applicationId
    val url = ssc.sparkContext.uiWebUrl.get
    val taskId = args(args.length-1)
    val register = args(args.length-2)

    val httpUrl = s"$register?taskId=$taskId&appId=$appId&state=REGISTER&url=$url"

    println(s"注册URL地址为:$httpUrl")

    val response = new DefaultHttpClient().execute(new HttpGet(httpUrl))
    if(response.getStatusLine.getStatusCode!=200){
      throw new Exception("注册失败")
    }

    println("*************** register end **************")

    ssc
  }

}

import com.alibaba.fastjson.JSON
import com.tgou.data.stanford.streaming.StreamingMonitor
import kafka.serializer.StringDecoder
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.kafka.KafkaUtils

object MonitoringTest {
  def main(args: Array[String]): Unit = {

    val ssc = StreamingMonitor.createContext(args,Seconds(5), "MonitoringTest")

    // Create direct kafka stream with brokers and topics
    val kafkaParams = Map[String, String](
      "metadata.broker.list" -> "hnode9:9092,hnode10:9092",
      "enable.auto.commit" -> "true")

    val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
      ssc, kafkaParams, Set("tgs-topic"))


    val lines = messages.map(_._2).map(line => JSON.parseObject(line))

    lines.print()

    ssc.start()

    ssc.awaitTermination()
  }
}

def main(args: Array[String]): Unit = {
    val ssc = StreamingContext.getOrCreate(CHECKPOINT_PATH,() => createContext(args))
    ssc.start()
    ssc.awaitTermination()
  }

  def createContext(args:Array[String]): StreamingContext = {


    /**************************************************************
    * 1. 创建StreamingContext，并完成注册的监控                      *
    * ************************************************************/

    val ssc = StreamingMonitor.createContext(args,WINDOW_TIME,MonitoringTest)
    ...