SparkStreaming DStream入门及其算子应用

什么是Dstream?

离散流（DStream）是Spark Streaming中的基本抽象，是表示相同数据流的RDD（相同类型）的连续序列。 DStreams可以使用StreamingContext从实时数据（例如来自TCP套接字，Kafka，Flume等的数据）创建，也可以通过使用map，Window和reduceByKeyAndWindow等操作转换现有DStreams来生成。在运行Spark Streaming程序时，每个DStream都会根据实时数据或通过转换父DStream生成的RDD定期生成RDD。

此类包含所有DStream上可用的基本操作，例如map，filter和window。另外，PairDStreamFunctions包含仅在键-值对的DStream上可用的操作，例如groupByKeyAndWindow和join。通过隐式转换，这些操作可在任何成对的DStream（例如DStream [（Int，Int）]）上自动使用。

DStream内部具有一些基本属性：
DStream依赖的其他DStream的列表
DStream生成RDD的时间间隔
每个时间间隔后用于生成RDD的函数

Dstream与RDD的关系

看源码可以知道,RRD结合时间尺度就是Dstream.

 private[streaming] var generatedRDDs = new HashMap[Time, RDD[T]]()

案例

DStream的算子相对RDD比较少,简单示例如下

以nc作为源发送数据

# nc -lk mypc01 10087

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}//sparkStreaming 算子测试案例
object Test extends App {private val conf = new SparkConf().setAppName("test").setMaster("local[*]").set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")private val ssc: StreamingContext = new StreamingContext(conf, Seconds(5))//以nc作为源转为Dstreamprivate val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("mypc01", 10087)//filetrDemo()//unionDemo()//countDemo()//countByvalue()//((o,6.0),2)//joinDemo()//(k,(1,2))//统计单词长度大于4的频率def filetrDemo(): Unit = {dstream.flatMap((_.split(" "))).filter(_.length > 4).map((_, 1)).reduceByKey(_ + _).print()}//测试union算子def unionDemo(): Unit = {val d1: DStream[(String, Int)] = dstream.map((_, 1))val d2: DStream[(String, Int)] = dstream.map((_, 1)).reduceByKey(_ + _)val d3: DStream[(String, Int)] = d1.union(d2)d3.print()}//测试count算子//统计每一批的个数def countDemo(): Unit = {val d1: DStream[Long] = dstream.count()d1.print()}//测试countByvalue算子def countByvalue(): Unit = {dstream.map((_, Math.floor(Math.random() * 10))).countByValue(2).print()}//测试join算子def joinDemo(): Unit = {val d1: DStream[(String, Int)] = dstream.map((_, 1))val d2: DStream[(String, Int)] = dstream.map((_, 1)).reduceByKey(_ + _)val d3: DStream[(String, (Int, Int))] = d1.join(d2)d3.print()}def cogroupDemo(): Unit = {val d1: DStream[(String, Int)] = dstream.map((_, 1))val d2: DStream[(String, Int)] = dstream.map((_, 1)).reduceByKey(_ + _)val d3: DStream[(String, (Iterable[Int], Iterable[Int]))] = d1.cogroup(d2)d3.print()}ssc.start()ssc.awaitTermination()
}

Dstream方法解析

Dstream应用map算子后返回值还是 Dstream类型,这个就好比Rdd应用map算子后返回值还是Rdd类型一样.

 def map[U: ClassTag](mapFunc: T => U): DStream[U] = ssc.withScope {new MappedDStream(this, context.sparkContext.clean(mapFunc))}

Dstream同样可以用filter算子

def filter(filterFunc: T => Boolean): DStream[T] = ssc.withScope {new FilteredDStream(this, context.sparkContext.clean(filterFunc))}

foreachRDD和上述的不同之处有亮点
1 没有返回值
2 参数为函数,且该函数的参数为Dstream所蕴含的RDD,这个就意味着内部也可以用RDD相关的算子.

def foreachRDD(foreachFunc: RDD[T] => Unit): Unit = ssc.withScope {val cleanedF = context.sparkContext.clean(foreachFunc, false)foreachRDD((r: RDD[T], _: Time) => cleanedF(r), displayInnerRDDOps = true)}

transform和foreachRDD是类似的,参数为函数,且该函数的参数是RDD类型

def transform[U: ClassTag](transformFunc: RDD[T] => RDD[U]): DStream[U] = ssc.withScope {// because the DStream is reachable from the outer object here, and because// DStreams can't be serialized with closures, we can't proactively check// it for serializability and so we pass the optional false to SparkContext.cleanval cleanedF = context.sparkContext.clean(transformFunc, false)transform((r: RDD[T], _: Time) => cleanedF(r))}

总结

Dstream就是RDD+时间尺度
注意理解foreachRDD算子以及transform算子与其他算子的不同之处