066 基于checkpoint的HA机制实现

 1 package com.stream.it
 2 
 3 import kafka.serializer.StringDecoder
 4 import org.apache.spark.storage.StorageLevel
 5 import org.apache.spark.streaming.kafka.KafkaUtils
 6 import org.apache.spark.streaming.{Seconds, StreamingContext}
 7 import org.apache.spark.{SparkConf, SparkContext}
 8 
 9 object HAKafkaWordcount {
10   def main(args: Array[String]): Unit = {
11     val conf=new SparkConf()
12         .setAppName("spark-streaming-wordcount")
13           .setMaster("local[*]")
14     val sc=SparkContext.getOrCreate(conf)
15     val checkpointDir = "hdfs://linux-hadoop01.ibeifeng.com:8020/beifeng/spark/streaming/chkdir02"
16 
17 
18     /**
19       * 构造StreamingContext对象
20       *
21       * @return
22       */
23     def createStreamingContextFunc(): StreamingContext = {
24       val ssc = new StreamingContext(sc, Seconds(5))
25       ssc.checkpoint(checkpointDir)
26       val kafkaParams=Map("group.id"->"stream-sparking-0",
27         "zookeeper.connect"->"linux-hadoop01.ibeifeng.com:2181/kafka",
28         "auto.offset.reset"->"smallest"
29       )
30       val topics=Map("beifeng"->1)
31       val dStream=KafkaUtils.createStream[String,String,StringDecoder,StringDecoder](
32         ssc,             //给定sparkStreaming的上下文
33         kafkaParams,     //kafka的参数信息，通过kafka HightLevelComsumerApi连接
34         topics,          //给定读取对应的topic的名称以及读取数据的线程数量
35         StorageLevel.MEMORY_AND_DISK_2     //数据接收器接收到kafka的数据后的保存级别
36       ).map(_._2)
37 
38 
39       val resultWordcount=dStream
40         .filter(line=>line.nonEmpty)
41         .flatMap(line=>line.split(" ").map((_,1)))
42         .reduceByKey(_+_)
43       resultWordcount.foreachRDD(rdd=>{
44         rdd.foreachPartition(iter=>iter.foreach(println))
45       })
46       ssc
47     }
48 
49     val ssc = StreamingContext.getOrCreate(
50       checkpointPath = checkpointDir,
51       creatingFunc = createStreamingContextFunc
52     )
53 
54     //启动
55     ssc.start()
56     //等到
57     ssc.awaitTermination()
58   }
59 }