Spark 学习笔记之 Streaming Window

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent

object WindowStreaming {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("KafkaDirect").setMaster("local[1]")
    val ssc = new StreamingContext(conf, Seconds(1))
    val kafkaMapParams = Map[String, Object](
      "bootstrap.servers" -> "192.168.1.151:9092,192.168.1.152:9092,192.168.1.153:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "g1",
      "auto.offset.reset" -> "latest", //earliest|latest
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )
    val topicsSet = Set("ScalaTopic")
    val kafkaStream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topicsSet, kafkaMapParams)
    )

    val finalResultRDD: DStream[(Int, String)] = kafkaStream.flatMap(row => row.value().split(" "))
      .map((_, 1)).reduceByKeyAndWindow((x: Int, y: Int) => x + y, Seconds(3), Seconds(2))
      .transform(rdd => rdd.map(tuple => (tuple._2, tuple._1))
        .sortByKey(false).map(tuple => (tuple._1, tuple._2))
      )

    finalResultRDD.print()

    ssc.start()
    ssc.awaitTermination()
  }


}