spark streaming窗口函数的使用和理解
spark streaming中的窗口函数虽然不如flink那么丰富,但是特别有用,看下面例子:
- kafkaStream.transform { rdd =>
- offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
- rdd}.map(_._2).map((_, 1)).reduceByKeyAndWindow((v1: Int, v2: Int) => {
- v1 + v2
- }, Seconds(8),
- 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,
看下面一个例子:
- package com.jingde.sparkstreamlast
- import kafka.serializer.StringDecoder
- import org.apache.log4j.{ Level, Logger }
- import org.apache.spark.SparkConf
- import org.apache.spark.rdd.RDD
- import org.apache.spark.streaming.kafka._
- import org.apache.spark.streaming.{ Seconds, StreamingContext }
- import org.apache.spark.streaming.kafka.KafkaUtils
- import org.apache.spark.streaming.kafka.OffsetRange
- import org.apache.log4j.{ Level, Logger }
- import org.I0Itec.zkclient.ZkClient
- import org.I0Itec.zkclient.exception.ZkMarshallingError
- import org.I0Itec.zkclient.serialize.ZkSerializer
- import kafka.utils.ZkUtils
- import kafka.utils.ZKGroupTopicDirs
- import org.apache.spark.streaming.dstream.InputDStream
- import kafka.common.TopicAndPartition
- import kafka.message.MessageAndMetadata
- import kafka.api.OffsetRequest
- import kafka.api.PartitionOffsetRequestInfo
- import kafka.consumer.SimpleConsumer
- import kafka.api.TopicMetadataRequest
- object StreamingFromKafka {
- val groupId = "logs"
- val topic = "streaming"
- val zkClient = new ZkClient("localhost:9999", 60000, 60000, new ZkSerializer {
- override def serialize(data: Object): Array[Byte] = {
- try {
- return data.toString().getBytes("UTF-8")
- } catch {
- case e: ZkMarshallingError => return null
- }
- }
- override def deserialize(bytes: Array[Byte]): Object = {
- try {
- return new String(bytes, "UTF-8")
- } catch {
- case e: ZkMarshallingError => return null
- }
- }
- })
- val topicDirs = new ZKGroupTopicDirs("spark_streaming_test", topic)
- val zkTopicPath = s"${topicDirs.consumerOffsetDir}"
- def main(args: Array[String]): Unit = {
- Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
- Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)
- val sparkConf = new SparkConf().setAppName("DirectKafkaWordCount")
- sparkConf.setMaster("local[*]")
- sparkConf.set("spark.streaming.kafka.maxRatePerPartition", "2")
- sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
- val ssc = new StreamingContext(sparkConf, Seconds(5))
- val kafkaParams = Map("metadata.broker.list" -> "localhost:9092", "group.id" -> groupId, "zookeeper.connect" -> "localhost:9999",
- "auto.offset.reset" -> kafka.api.OffsetRequest.SmallestTimeString)
- val topics = Set(topic)
- val children = zkClient.countChildren(s"${topicDirs.consumerOffsetDir}")
- var kafkaStream: InputDStream[(String, String)] = null
- var fromOffsets: Map[TopicAndPartition, Long] = Map()
- ssc.checkpoint("D:\tmp\storm-hdfs") //这里是hdfs路径,因为要做窗口函数,需要用到前面的rdd,这里必须要用checkpoint
- if (children > 0) {
- //---get partition leader begin----
- val topicList = List(topic)
- val req = new TopicMetadataRequest(topicList, 0) //得到该topic的一些信息,比如broker,partition分布情况
- val getLeaderConsumer = new SimpleConsumer("localhost", 9092, 10000, 10000, "OffsetLookup") // brokerList的host 、brokerList的port、过期时间、过期时间
- val res = getLeaderConsumer.send(req) //TopicMetadataRequest topic broker partition 的一些信息
- val topicMetaOption = res.topicsMetadata.headOption
- val partitions = topicMetaOption match {
- case Some(tm) =>
- tm.partitionsMetadata.map(pm => (pm.partitionId, pm.leader.get.host)).toMap[Int, String]
- case None =>
- Map[Int, String]()
- }
- for (i <- 0 until children) {
- val partitionOffset = zkClient.readData[String](s"${topicDirs.consumerOffsetDir}/${i}")
- val tp = TopicAndPartition(topic, i)
- //---additional begin-----
- val requestMin = OffsetRequest(Map(tp -> PartitionOffsetRequestInfo(OffsetRequest.EarliestTime, 1))) // -2,1
- val consumerMin = new SimpleConsumer(partitions(i), 9092, 10000, 10000, "getMinOffset")
- val curOffsets = consumerMin.getOffsetsBefore(requestMin).partitionErrorAndOffsets(tp).offsets
- var nextOffset = partitionOffset.toLong
- if (curOffsets.length > 0 && nextOffset < curOffsets.head) { //如果下一个offset小于当前的offset
- nextOffset = curOffsets.head
- }
- //---additional end-----
- fromOffsets += (tp -> nextOffset)
- fromOffsets += (tp -> partitionOffset.toLong) //将不同 partition 对应的 offset 增加到 fromOffsets 中
- }
- val messageHandler = (mmd: MessageAndMetadata[String, String]) => (mmd.topic, mmd.message()) //这个会将 kafka 的消息进行 transform,最终 kafak 的数据都会变成 (topic_name, message) 这样的 tuple
- kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String)](ssc, kafkaParams, fromOffsets, messageHandler)
- } else {
- kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)
- }
- var offsetRanges = Array[OffsetRange]()
- kafkaStream.transform { rdd =>
- offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
- rdd
- }.map(_._2).map((_, 1)).reduceByKeyAndWindow((v1: Int, v2: Int) => {
- v1 + v2
- }, Seconds(5), //每隔5秒(后面的5秒),计算最近5秒(前面的5秒)的数据
- Seconds(5)).foreachRDD {
- rdd =>
- rdd.foreachPartition { element => element.foreach {
- println } }
- for (o <- offsetRanges) {
- ZkUtils.updatePersistentPath(zkClient, s"${topicDirs.consumerOffsetDir}/${o.partition}", o.fromOffset.toString)
- }
- }
- ssc.start()
- ssc.awaitTermination()
- }
- }
相关推荐
Kafka 2020-09-18
Wepe0 2020-10-30
杜倩 2020-10-29
windle 2020-10-29
minerd 2020-10-28
mengzuchao 2020-10-22
Junzizhiai 2020-10-10
bxqybxqy 2020-09-30
风之沙城 2020-09-24
kingszelda 2020-09-22
大唐帝国前营 2020-08-18
yixu0 2020-08-17
TangCuYu 2020-08-15
xiaoboliu00 2020-08-15
songshijiazuaa 2020-08-15
xclxcl 2020-08-03
zmzmmf 2020-08-03
newfarhui 2020-08-03
likesyour 2020-08-01