package com.window.it

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

 import org.apache.spark.storage.StorageLevel

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

 import org.apache.spark.streaming.dstream.DStream

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

 object TransformDemo {

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

     val conf = new SparkConf()

       .setAppName("StreamingWindowOfKafka")

       .setMaster("local[*]")

     val sc = SparkContext.getOrCreate(conf)

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

     // 当调用updateStateByKey函数API的时候，必须给定checkpoint dir

     // 路径对应的文件夹不能存在

     ssc.checkpoint("hdfs://linux-hadoop01.ibeifeng.com:8020/beifeng/spark/streaming/4525712")

     val kafkaParams = Map(

       "group.id" -> "streaming-kafka-78912151",

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

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

     )

     val topics = Map("beifeng" -> 4) // topics中value是读取数据的线程数量，所以必须大于等于1

     val dstream = KafkaUtils.createStream[String, String, kafka.serializer.StringDecoder, kafka.serializer.StringDecoder](

       ssc, // 给定SparkStreaming上下文

       kafkaParams, // 给定连接kafka的参数信息 ===> 通过Kafka HighLevelConsumerAPI连接

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

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

     ).map(_._2)

     val resultWordCount = dstream

       .filter(line => line.nonEmpty)

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

       .reduceByKeyAndWindow(

         (a: Int, b: Int) => a + b,

         Seconds(15), // 窗口大小

         Seconds(10) // 滑动大小

       )

     resultWordCount.print() // 这个也是打印数据

     /**

       * transform：将DStream的操作转换为RDD的操作，调用该api最终只需要返回一个新的RDD即可

       */

     dstream.transform(rdd => {

       // 对rdd进行预处理

       val processedRDD = rdd

         .filter(line => line.nonEmpty)

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

         .reduceByKey(_ + _)

       // 数据抽样，获取两个节点

       val seeder = processedRDD.takeSample(true, 2)

       // 对rdd进行处理操作, 将抽样数据和rdd中的数据进行比较，如果rdd中的word的出现次数大于等于抽样数据中的任何一个word的次数，次数*3；否则次数*2

       val brocast = rdd.sparkContext.broadcast(seeder)

       val resultRDD = processedRDD.mapPartitions(iter => {

         val seederValue = brocast.value

         iter.map {

           case (word, count) => {

             val vc = seederValue

               .filter(tuple => {

                 count >= tuple._2

               }).size

             if (vc == 0) {

               (word, 2, count * 2)

             } else {

               (word, 3, count * 3)

             }

           }

         }

       })

       resultRDD

     }).print()

     // 启动开始处理

     ssc.start()

     ssc.awaitTermination() // 等等结束，监控一个线程的中断操作

   }

 }

 package com.window.it

 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 TransformDemo {

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

     val conf = new SparkConf()

       .setAppName("StreamingWindowOfKafka")

       .setMaster("local[*]")

     val sc = SparkContext.getOrCreate(conf)

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

     // 当调用updateStateByKey函数API的时候，必须给定checkpoint dir

     // 路径对应的文件夹不能存在

     ssc.checkpoint("hdfs://linux-hadoop01.ibeifeng.com:8020/beifeng/spark/streaming/4525712")

     val kafkaParams = Map(

       "group.id" -> "streaming-kafka-78912151",

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

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

     )

     val topics = Map("beifeng" -> 4) // topics中value是读取数据的线程数量，所以必须大于等于1

     val dstream = KafkaUtils.createStream[String, String, kafka.serializer.StringDecoder, kafka.serializer.StringDecoder](

       ssc, // 给定SparkStreaming上下文

       kafkaParams, // 给定连接kafka的参数信息 ===> 通过Kafka HighLevelConsumerAPI连接

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

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

     ).map(_._2)

     val resultWordCount = dstream

       .filter(line => line.nonEmpty)

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

       .reduceByKeyAndWindow(

         (a: Int, b: Int) => a + b,

         Seconds(15), // 窗口大小

         Seconds(10) // 滑动大小

       )

     resultWordCount.print() // 这个也是打印数据

     dstream.foreachRDD(rdd => {

       // TODO: 这里就可以做数据输出的代码编写

       // TODO: 这里不要为空

       rdd.foreachPartition(iter => {

         // TODO: 这里在实际环境中不要为空，为空可能会出现一些问题：内存泄露的问题

         println(iter.take(1))

       })

     })

     // 启动开始处理

     ssc.start()

     ssc.awaitTermination() // 等等结束，监控一个线程的中断操作

   }

 }