2. 运算

2.1. StreamingContext.checkPoint

    val ssc = new StreamingContext(sc, Seconds(10))val checkpointDir = "hdfs://dir/checkpoint"ssc.checkpoint(checkpointDir)

1、为Spark Streaming设置checkpoint，在使用DStream.updateByKey前必须先设置checkpoint
2、在checkpointDir目录下会生成形如checkpoint-1454403510000和checkpoint-1454403510000.bk的文件和1f355a78-5404-4b47-9755-8529d15e9037的文件夹
3、checkpoint-1454403510000和checkpoint-1454403510000.bk内容一样，很明显1454403510000是时间戳，打开可以看见形如update hdfs://dir/1f355a78-5404-4b47-9755-8529d15e9037/rdd-87的语句，说明把当前状态存入了rdd-87中

2.2. DStream.updateByKey

    val streamPath = "hdfs://dir/1"val words: DStream[(String, Int)] = ssc.textFileStream(streamPath).flatMap(_.split(" ")).map(x => (x, 1))//为每个word转换为(word,1)的tupleval wordCounts = words.updateStateByKey[Int](updateFunc = (sumNow: Seq[Int], sumLast: Option[Int]) => {Some(sumNow.sum + sumLast.getOrElse(0))})wordCounts.print()ssc.start()ssc.awaitTermination()

updateByKey的官网解释
1、updateByKey是PairDStreamFunctions的算子，必须是DStream[(Key, Value)]才能使用。
2、updateByKey最长接受的参数如下：

    def updateStateByKey[S](updateFunc: (Iterator[(K, Seq[V], Option[S])]) ⇒ Iterator[(K, S)], partitioner: Partitioner, rememberPartitioner: Boolean, initialRDD: RDD[(K, S)])(implicit arg0: ClassTag[S]): DStream[(K, S)]

最短的如下：

    def updateStateByKey[S](updateFunc: (Seq[V], Option[S]) ⇒ Option[S])(implicit arg0: ClassTag[S]): DStream[(K, S)]

其中最重要的是updateFunc: (Seq[V], Option[S]) ⇒ Option[S])，它来说明如何更新状态(state)

    val wordCounts = words.updateStateByKey[Int](updateFunc = (sumNow: Seq[Int], sumLast: Option[Int]) => {Some(sumNow.sum + sumLast.getOrElse(0))})

1、假设words中是形如DStream((“hello”,1),(“hello”,1),(“hello”,1),(“hello”,1),(“hello”,1),(“spark”,1),(“spark”,1)）
2、对于Key = “hello”来说updateFunc的输入参数sumNow就是Seq(1,1,1,1,1)，sumLast是上个状态”hello”的Value值——空值；输出就是本状态”hello”的Value值——Some(1+1+1+1+1+0) = Some(5)。
3、最终wordCounts就是形如DStream((“hello”,5),(“spark”,2))

2.2.1. 期望只统计最近2个state的数据

val input = ssc.textFileStream(streamPath)
val words = input.flatMap(_.split(" ")).map(x => (x, Array(1,0)))val wordCounts = words.updateStateByKey[Array[Int]](updateFunc = (countNow: Seq[Array[Int]], stateLast: Option[Array[Int]]) => {val sumNow: Int = countNow.map(_(0)).sumval sumLast: Int = stateLast.getOrElse(Array(0,0))(0)Some(Array(sumNow,sumLast))})wordCounts.map(x => (x._1, x._2.sum)).print()

2.2.2. 期望定义state’有新文件才生成新的state’，而不是目前每10秒一个新的state

1、尚未发现有API支持这一点
2、尚未发现有方法判定当前DStream为空的API
3、尚未发现有API可将DStream => 除了DStream、Unit和StreamingContext之外的数据结构
4、尝试采用如下语句判定：

    var flag = trueval input = ssc.textFileStream(streamPath)input.foreachRDD(x => {if (x.isEmpty()) flag = false else flag = true})val words = input.flatMap(_.split(" ")).map(x => (x, Array(1,0)))val wordCounts = words.updateStateByKey[Array[Int]](updateFunc = (countNow: Seq[Array[Int]], stateLast: Option[Array[Int]]) => {if (flag) {val sumNow: Int = countNow.map(_(0)).sumval sumLast: Int = stateLast.getOrElse(Array(0,0))(0)Some(Array(sumNow,sumLast))} else {stateLast}})wordCounts.map(x => (x._1, x._2.sum)).print()

结果Spark-shell中报错：
org.apache.spark.SparkException: Task not serializable
Caused by: java.io.NotSerializableException: org.apache.spark.streaming.StreamingContext
2016.2.17日更新：Spark-submit成功！

2.3. DStream.reduceByKeyAndWindow

业务场景：
1、约每5分钟生成一个源文件放在hdfs://dir中，该文件代表1分钟内收到的所有单词
2、统计当前所有单词的出现次数
3、该单词如果40分钟内未出现，既40个文件中未出现则认为消失，不出现在统计结果中
4、要求每10秒扫描一次文件夹
由于2.2.2的问题没有解决，也就是说Spark Streaming每10秒一个state，而不是每个新文件一个state。由于累加结果有40分钟过期时间，因此无法用单纯的updateByKey算子来计算。尝试采用DStream.reduceByKeyAndWindow算子来替代。
采用2.2.2的方式实现

Spark Streaming学习与实践(2)相关推荐

Spark学习笔记(8)---Spark Streaming学习笔记
Spark Streaming学习笔记同Spark SQL一样,Spark Streaming学习也是放在了github https://github.com/yangtong123/RoadOfS ...
Spark大数据分析与实战：Spark Streaming编程初级实践
Spark Streaming编程初级实践一.安装Hadoop和Spark 具体的安装过程在我以前的博客里面有,大家可以通过以下链接进入操作: Hadoop的安装:https://blog.csdn ...
pythonspark实践_基于Python的Spark Streaming Kafka编程实践
版权声明:本文为CSDN博主原创文章,未经博主允许不得转载. 说明 Spark Streaming的原理说明的文章很多,这里不做介绍.本文主要介绍使用Kafka作为数据源的编程模型,编码实践,以及一些 ...
Spark Streaming学习笔记
特点: Spark Streaming能够实现对实时数据流的流式处理,并具有很好的可扩展性.高吞吐量和容错性. Spark Streaming支持从多种数据源提取数据,如:Kafka.Flume.Tw ...
Spark Streaming实践和优化
2019独角兽企业重金招聘Python工程师标准>>> Spark Streaming实践和优化博客分类: spark 在流式计算领域,Spark Streaming和Storm时 ...
Spark Streaming高级特性在NDCG计算实践
从storm到spark streaming,再到flink,流式计算得到长足发展, 依托于spark平台的spark streaming走出了一条自己的路,其借鉴了spark批处理架构,通过批处理方 ...
StreamDM：基于Spark Streaming、支持在线学习的流式分析算法引擎
StreamDM:基于Spark Streaming.支持在线学习的流式分析算法引擎 streamDM:Data Mining for Spark Streaming,华为诺亚方舟实验室开源了业界第一 ...
Spark学习之Spark Streaming（9）
Spark学习之Spark Streaming(9) 1. Spark Streaming允许用户使用一套和批处理非常接近的API来编写流式计算应用,这就可以大量重用批处理应用的技术甚至代码. 2. ...
大数据学习15之spark streaming入门
文章目录一.概述二.应用场景三.集成Spark生态系统的使用四.发展史五.从词频统计功能着手入门 1.spark-submit执行 2.spark-shell执行(测试时使用) 六.工作原理 ...

Spark Streaming学习与实践(2)