package com.stream.it

 import kafka.serializer.StringDecoder

 import org.apache.spark.storage.StorageLevel

 import org.apache.spark.streaming.kafka.KafkaUtils

 import org.apache.spark.streaming.{Seconds, StreamingContext}

 import org.apache.spark.{SparkConf, SparkContext}

 object HAKafkaWordcount {

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

     val conf=new SparkConf()

         .setAppName("spark-streaming-wordcount")

           .setMaster("local[*]")

     val sc=SparkContext.getOrCreate(conf)

     val checkpointDir = "hdfs://linux-hadoop01.ibeifeng.com:8020/beifeng/spark/streaming/chkdir02"

     /**

       * 构造StreamingContext对象

       *

       * @return

       */

     def createStreamingContextFunc(): StreamingContext = {

       val ssc = new StreamingContext(sc, Seconds(5))

       ssc.checkpoint(checkpointDir)

       val kafkaParams=Map("group.id"->"stream-sparking-0",

         "zookeeper.connect"->"linux-hadoop01.ibeifeng.com:2181/kafka",

         "auto.offset.reset"->"smallest"

       )

       val topics=Map("beifeng"->1)

       val dStream=KafkaUtils.createStream[String,String,StringDecoder,StringDecoder](

         ssc,             //给定sparkStreaming的上下文

         kafkaParams,     //kafka的参数信息，通过kafka HightLevelComsumerApi连接

         topics,          //给定读取对应的topic的名称以及读取数据的线程数量

         StorageLevel.MEMORY_AND_DISK_2     //数据接收器接收到kafka的数据后的保存级别

       ).map(_._2)

       val resultWordcount=dStream

         .filter(line=>line.nonEmpty)

         .flatMap(line=>line.split(" ").map((_,1)))

         .reduceByKey(_+_)

       resultWordcount.foreachRDD(rdd=>{

         rdd.foreachPartition(iter=>iter.foreach(println))

       })

       ssc

     }

     val ssc = StreamingContext.getOrCreate(

       checkpointPath = checkpointDir,

       creatingFunc = createStreamingContextFunc

     )

     //启动

     ssc.start()

     //等到

     ssc.awaitTermination()

   }

 }