spark streaming窗口函数的使用和理解

spark streaming中的窗口函数虽然不如flink那么丰富，但是特别有用，看下面例子：

1. 
   
2. kafkaStream.transform { rdd =>
3. offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
4. rdd}.map(_._2).map((_, 1)).reduceByKeyAndWindow((v1: Int, v2: Int) => {
5. v1 + v2
6. }, Seconds(8),
7. Seconds(4))

表示每隔4秒(后面的4秒)，计算最近8秒(前面的8秒)的数据。

第一个时间称为窗口长度，第二个时间称为滑动长度，其含义表示每隔4秒计算最近最近8秒的数据，这可以用于一些业务场景，例如网站记录，每隔1个小时计算最近两个小时的pv量，还有一种业务场景的话先在内存中做累加再更新到redis中做累加，比如说每隔5秒统计最近5秒的数据的总和，再刷到redis中做累加，因为频繁操作redis的话会存在问题，还有一个时间如下：

val ssc = new StreamingContext(sparkConf, Seconds(4))

此处设置的batch Interval是在spark streaming中生成基本Job的时间单位，窗口和滑动时间间隔一定是该batch Interval的整数倍，若要在内存中做简单的累加只要设置窗口长度和滑动长度相同即可。

持久化：

因为要窗口函数要用前面所用到的rdd，在这里必须checkpoint，

看下面一个例子：

1. 
   
2. package com.jingde.sparkstreamlast
3. import kafka.serializer.StringDecoder
4. import org.apache.log4j.{ Level, Logger }
5. import org.apache.spark.SparkConf
6. import org.apache.spark.rdd.RDD
7. import org.apache.spark.streaming.kafka._
8. import org.apache.spark.streaming.{ Seconds, StreamingContext }
9. import org.apache.spark.streaming.kafka.KafkaUtils
10. import org.apache.spark.streaming.kafka.OffsetRange
11. import org.apache.log4j.{ Level, Logger }
12. import org.I0Itec.zkclient.ZkClient
13. import org.I0Itec.zkclient.exception.ZkMarshallingError
14. import org.I0Itec.zkclient.serialize.ZkSerializer
15. import kafka.utils.ZkUtils
16. import kafka.utils.ZKGroupTopicDirs
17. import org.apache.spark.streaming.dstream.InputDStream
18. import kafka.common.TopicAndPartition
19. import kafka.message.MessageAndMetadata
20. import kafka.api.OffsetRequest
21. import kafka.api.PartitionOffsetRequestInfo
22. import kafka.consumer.SimpleConsumer
23. import kafka.api.TopicMetadataRequest
24. object StreamingFromKafka {
25. val groupId = "logs"
26. val topic = "streaming"
27. val zkClient = new ZkClient("localhost:9999", 60000, 60000, new ZkSerializer {
28. override def serialize(data: Object): Array[Byte] = {
29. try {
30. return data.toString().getBytes("UTF-8")
31. } catch {
32. case e: ZkMarshallingError => return null
33. }
34. }
35. override def deserialize(bytes: Array[Byte]): Object = {
36. try {
37. return new String(bytes, "UTF-8")
38. } catch {
39. case e: ZkMarshallingError => return null
40. }
41. }
42. })
43. val topicDirs = new ZKGroupTopicDirs("spark_streaming_test", topic)
44. val zkTopicPath = s"${topicDirs.consumerOffsetDir}"
45. def main(args: Array[String]): Unit = {
46. Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
47. Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)
48. val sparkConf = new SparkConf().setAppName("DirectKafkaWordCount")
49. sparkConf.setMaster("local[*]")
50. sparkConf.set("spark.streaming.kafka.maxRatePerPartition", "2")
51. sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
52. val ssc = new StreamingContext(sparkConf, Seconds(5))
53. val kafkaParams = Map("metadata.broker.list" -> "localhost:9092", "group.id" -> groupId, "zookeeper.connect" -> "localhost:9999",
54. "auto.offset.reset" -> kafka.api.OffsetRequest.SmallestTimeString)
55. val topics = Set(topic)
56. val children = zkClient.countChildren(s"${topicDirs.consumerOffsetDir}")
57. var kafkaStream: InputDStream[(String, String)] = null
58. var fromOffsets: Map[TopicAndPartition, Long] = Map()
59. ssc.checkpoint("D:\tmp\storm-hdfs") //这里是hdfs路径，因为要做窗口函数，需要用到前面的rdd，这里必须要用checkpoint
60. if (children > 0) {
61. //---get partition leader begin----
62. val topicList = List(topic)
63. val req = new TopicMetadataRequest(topicList, 0) //得到该topic的一些信息，比如broker,partition分布情况
64. val getLeaderConsumer = new SimpleConsumer("localhost", 9092, 10000, 10000, "OffsetLookup") // brokerList的host 、brokerList的port、过期时间、过期时间
65. val res = getLeaderConsumer.send(req) //TopicMetadataRequest topic broker partition 的一些信息
66. val topicMetaOption = res.topicsMetadata.headOption
67. val partitions = topicMetaOption match {
68. case Some(tm) =>
69. tm.partitionsMetadata.map(pm => (pm.partitionId, pm.leader.get.host)).toMap[Int, String]
70. case None =>
71. Map[Int, String]()
72. }
73. for (i <- 0 until children) {
74. val partitionOffset = zkClient.readData[String](s"${topicDirs.consumerOffsetDir}/${i}")
75. val tp = TopicAndPartition(topic, i)
76. //---additional begin-----
77. val requestMin = OffsetRequest(Map(tp -> PartitionOffsetRequestInfo(OffsetRequest.EarliestTime, 1))) // -2,1
78. val consumerMin = new SimpleConsumer(partitions(i), 9092, 10000, 10000, "getMinOffset")
79. val curOffsets = consumerMin.getOffsetsBefore(requestMin).partitionErrorAndOffsets(tp).offsets
80. var nextOffset = partitionOffset.toLong
81. if (curOffsets.length > 0 && nextOffset < curOffsets.head) { //如果下一个offset小于当前的offset
82. nextOffset = curOffsets.head
83. }
84. //---additional end-----
85. fromOffsets += (tp -> nextOffset)
86. fromOffsets += (tp -> partitionOffset.toLong) //将不同 partition 对应的 offset 增加到 fromOffsets 中
87. }
88. val messageHandler = (mmd: MessageAndMetadata[String, String]) => (mmd.topic, mmd.message()) //这个会将 kafka 的消息进行 transform，最终 kafak 的数据都会变成 (topic_name, message) 这样的 tuple
89. kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String)](ssc, kafkaParams, fromOffsets, messageHandler)
90. } else {
91. kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)
92. }
93. var offsetRanges = Array[OffsetRange]()
94. kafkaStream.transform { rdd =>
95. offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
96. rdd
97. }.map(_._2).map((_, 1)).reduceByKeyAndWindow((v1: Int, v2: Int) => {
98. v1 + v2
99. }, Seconds(5), //每隔5秒(后面的5秒)，计算最近5秒(前面的5秒)的数据
100. Seconds(5)).foreachRDD {
101. rdd =>
102. rdd.foreachPartition { element => element.foreach {
104. println } }
105. for (o <- offsetRanges) {
106. ZkUtils.updatePersistentPath(zkClient, s"${topicDirs.consumerOffsetDir}/${o.partition}", o.fromOffset.toString)
107. }
108. }
109. ssc.start()
110. ssc.awaitTermination()
111. }
112. }