spark-streaming 编程(六)mapwithState

mapWithState的用法

message.mapWithState(StateSpec.function(func).initialState(RDD).timeout(time))

需要自己写一个匿名函数func来实现自己想要的功能。如果有初始化的值得需要，可以使用initialState(RDD)来初始化key的值。
另外，还可以指定timeout函数，该函数的作用是，如果一个key超过timeout设定的时间没有更新值，那么这个key将会失效。这个控制需要在func中实现，必须使用state.isTimingOut()来判断失效的key值。如果在失效时间之后，这个key又有新的值了，则会重新计算。如果没有使用isTimingOut，则会报错。

代码示例：

package com.lgh.sparkstreamingimport kafka.serializer.StringDecoder
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, State, StateSpec, StreamingContext}
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka.KafkaUtilsimport scala.collection.mutable.Set/*** Created by lgh on 2017/8/24.*/
object mapWithState {def main(args: Array[String]): Unit = {val brokers = "mtime-bigdata00:9092,mtime-bigdata01:9092";val topics = "testkafka";val batchseconds = "10";val checkpointDirectory = "./mapwithstate";val ssc = StreamingContext.getOrCreate(checkpointDirectory, () => createcontext(brokers, topics, batchseconds, checkpointDirectory))ssc.start()ssc.awaitTermination()}def createcontext(brokers: String, topics: String, batchseconds: String, checkpointDirectory: String): StreamingContext = {val sparkconf = new SparkConf().setAppName("TestUpStateByKey").setMaster("local[3]")val ssc = new StreamingContext(sparkconf, Seconds(batchseconds.toInt))val topicsSet = topics.split(",").toSetval kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)val lines: DStream[String] = messages.map(_._2)val message: DStream[(String, Long)] = lines.flatMap(_.split(" ")).map(x=>(x,1l)).reduceByKey(_+_)//匿名函数val logical = (key: String, value: Option[Long], state: State[Long])=>{//这个的作用是检测已经过期的key并移除；如果有key过期后又有这个key新的数据进来，不加isTimeout的话就会导致报错if(state.isTimingOut()){System.out.println(key+" is timingout")}else {val sum = state.getOption().getOrElse(0l) + value.getOrElse(0l)val output = (key, sum)//更新状态state.update(sum)output}}val keyvalue=message.mapWithState(StateSpec.function(logical).timeout(Seconds(60)))keyvalue.stateSnapshots().foreachRDD((rdd,time)=>{println("========"+rdd.count())rdd.foreach( x=>println(x._1+"="+x._2))})ssc.checkpoint(checkpointDirectory)ssc}}

spark-streaming 编程(六)mapwithState相关推荐

编程实现将rdd转换为dataframe:源文件内容如下(_第四篇|Spark Streaming编程指南(1)
Spark Streaming是构建在Spark Core基础之上的流处理框架,是Spark非常重要的组成部分.Spark Streaming于2013年2月在Spark0.7.0版本中引入,发展至今 ...
Spark Streaming 编程指南[中英对照]
2019独角兽企业重金招聘Python工程师标准>>> 基于Spark 2.0 Preview的材料翻译,原[英]文地址: http://spark.apache.org/docs/ ...
Spark大数据分析与实战：Spark Streaming编程初级实践
Spark Streaming编程初级实践一.安装Hadoop和Spark 具体的安装过程在我以前的博客里面有,大家可以通过以下链接进入操作: Hadoop的安装:https://blog.csdn ...
Spark Streaming 编程新手入门指南
Spark Streaming 是核心Spark API的扩展,可实现实时数据流的可伸缩,高吞吐量,容错流处理.可以从许多数据源(例如Kafka,Flume,Kinesis或TCP sockets)中 ...
sparkstreaming监听hdfs目录如何终止_四十六、Spark Streaming简介及入门
1.什么是Spark Streaming Spark Streaming是基于Spark Core之间的实时计算框架,可以从很多数据源消费数据并对数据进行处理.它是Spark核心API的一个扩展与封装 ...
基于大数据的Uber数据实时监控（Part 2：Kafka和Spark Streaming）
导言本文是系列文章的第二篇,我们将建立一个分析和监控Uber汽车GPS旅行数据的实时示例.在第一篇文章中讨论了使用Apache Spark的K-means算法创建机器学习模型,以根据位置聚类Uber ...
hive编程指南电子版_第三篇|Spark SQL编程指南
在<第二篇|Spark Core编程指南>一文中,对Spark的核心模块进行了讲解.本文将讨论Spark的另外一个重要模块--Spark SQL,Spark SQL是在Shark的基础之上 ...
Spark详解（十二）：Spark Streaming原理和实现
1 简介 SparkStreaming是Spark核心API的一个扩展,具有高吞吐量和容错能力的实时流数据处理系统,可以对多种数据源(如Kdfka.Flume.Twitter.Zero和TCP 套接字 ...
spark-sql建表语句限制_第三篇|Spark SQL编程指南
在<第二篇|Spark Core编程指南>一文中,对Spark的核心模块进行了讲解.本文将讨论Spark的另外一个重要模块--Spark SQL,Spark SQL是在Shark的基础之上 ...
Spark Streaming实时数据分析
1.Spark Streaming功能介绍 1)定义 Spark Streaming is an extension of the core Spark API that enables scalab ...

spark-streaming 编程(六)mapwithState

spark-streaming 编程(六)mapwithState相关推荐

最新文章

热门文章