spark Streaming

Posted 2022-01-14 hejunhong

Spark Streaming输入

 

//*************** Spark Streaming 的输入 **************

 

1、 文件数据源

     1、Spark Streaming通过streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory) 这个方法提供了对目录下文件数据源的支持。

      2、如果你的文件是比较简单的文本文件，你可以使用 streamingContext.textFileStream(dataDirectory)  来代替。

        3、文件数据源目前不支持嵌套目录：

                     1、 文件需要有相同的数据格式

         2、文件进入 dataDirectory的方式需要通过移动或者重命名来实现

         3、一旦文件移动进目录，则不能再修改，即便修改了也不会读取新数据。

  

2、自定义Receiver

    1、你需要新建一个Class去继承Receiver，并给Receiver传入一个类型参数，该类型参数是你需要接收的数据的类型。

    2、你需要去复写Receiver的方法： onStart方法（在Receiver启动的时候调用的方法）、onStop方法（在Receiver正常停止的情况下调用的方法）

    3、你可以在程序中通过streamingContext.receiverStream( new CustomeReceiver)来调用你定制化的Receiver。

 

3、RDD数据源

     1、你可以通过StreamingContext.queueStream(rddQueue)这个方法来监控一个RDD的队列，所有加入到这个RDD队列中的新的RDD，都会被Streaming去处理。

 

 

4、Spark Streaming和Kafka的集成

     

================================================

Spark Streaming输出

 

//************** Spark Streaming 的输出 ****************

print()

在运行流程序的驱动结点上打印DStream中每一批次数据的最开始10个元素。这用于开发和调试。在Python API中，同样的操作叫pprint()。 

 

saveAsTextFiles(prefix, [suffix])

以text文件形式存储这个DStream的内容。每一批次的存储文件名基于参数中的prefix和suffix。”prefix-Time_IN_MS[.suffix]”. 

 

saveAsObjectFiles(prefix, [suffix])

以Java对象序列化的方式将Stream中的数据保存为 SequenceFiles . 每一批次的存储文件名基于参数中的为"prefix-TIME_IN_MS[.suffix]". Python中目前不可用。

saveAsHadoopFiles(prefix, [suffix])

将Stream中的数据保存为 Hadoop files. 每一批次的存储文件名基于参数中的为"prefix-TIME_IN_MS[.suffix]". 

Python API Python中目前不可用。

foreachRDD(func)

这是最通用的输出操作，即将函数func用于产生于stream的每一个RDD。其中参数传入的函数func应该实现将每一个RDD中数据推送到外部系统，如将RDD存入文件或者通过网络将其写入数据库。注意：函数func在运行流应用的驱动中被执行，同时其中一般函数RDD操作从而强制其对于流RDD的运算。

 

Spark Streaming重点知识

 

//************* Spark Streaming是什么

1、SPark Streaming是Spark中一个组件，基于Spark Core进行构建，用于对流式进行处理，类似于Storm。

2、Spark Streaming能够和Spark Core、Spark SQL来进行混合编程。

3、Spark Streaming我们主要关注：

      1、Spark Streaming 能接受什么数据？ kafka、flume、HDFS、Twitter等。

      2、Spark Streaming 能怎么处理数据？ 无状态的转换（前面处理的数据和后面处理的数据没啥关系）、有转换转换（前面处理的数据和后面处理的数据是有关系的，比如叠加关系）

 

//************* Spark Streaming是怎么实现的

1、Spark Streaming 采用“微批次”架构。

2、对于整个流式计算来说，数据流你可以想象成水流，微批次架构的意思就是将水流按照用户设定的时间间隔分割为多个水流段。一个段的水会在Spark中转换成为一个RDD，所以对水流的操作也就是对这些分割后的RDD进行单独的操作。每一个RDD的操作都可以认为是一个小的批处理（也就是离线处理）。

 

//************* Spark Streaming DStream是啥

1、DStream是类似于RDD和DataFrame的针对流式计算的抽象类。在源码中DStream是通过HashMap来保存他所管理的数据流的。K是RDD中数据流的时间，V是包含数据流的RDD。

2、对于DStream的操作也就是对于DStream他所包含的所有以时间序列排序的RDD的操作。

 

//************* Spark Streaming 怎么用

1、通过StreamingContext来进入Spark Streaming。可以通过已经创建好的SparkContext来创建SparkStreaming。

 

Spark Streaming转换

 

//************ Spark Streaming的转换 ******************

 

1、这些方法定义在哪儿？

     1、DStream的类定义中，主要提供对于值类型的DStream的操作。

     2、PairDStreamFunctions的类定义中，主要提供对于K-V形式的DStream的操作。

 

2、无状态装换

     1、def map[U: ClassTag](mapFunc: T => U): DStream[U] 将源DStream中的每个元素通过一个函数func从而得到新的DStreams。

    2、def flatMap[U: ClassTag](flatMapFunc: T => TraversableOnce[U]): DStream[U] 和map类似，但是每个输入的项可以被映射为0或更多项。

    3、def filter(filterFunc: T => Boolean): DStream[T] 选择源DStream中函数func判为true的记录作为新DStream

        4、def repartition(numPartitions: Int): DStream[T]   通过创建更多或者更少的partition来改变此DStream的并行级别。

    5、def union(that: DStream[T]): DStream[T]  将一个具有相同slideDuration新的DStream和当前DStream进行合并，返回新的DStream

    6、def count(): DStream[Long]  统计源DStreams中每个RDD所含元素的个数得到单元素RDD的新DStreams。

    7、def reduce(reduceFunc: (T, T) => T): DStream[T]  通过函数func(两个参数一个输出)来整合源DStreams中每个RDD元素得到单元素RDD的DStream。

        8、def countByValue(numPartitions: Int = ssc.sc.defaultParallelism)(implicit ord: Ordering[T] = null)

    : DStream[(T, Long)]   对于DStreams中元素类型为K调用此函数，得到包含(K,Long)对的新DStream，其中Long值表明相应的K在源DStream中每个RDD出现的次数。

    9、def reduceByKey(reduceFunc: (V, V) => V): DStream[(K, V)]  对(K,V)对的DStream调用此函数，返回同样（K,V)对的新DStream，但是新DStream中的对应V为使用reduce函数整合而来

    10、def join[W: ClassTag](other: DStream[(K, W)]): DStream[(K, (V, W))]  两DStream分别为(K,V)和(K,W)对，返回(K,(V,W))对的新DStream。

    11、def cogroup[W: ClassTag](other: DStream[(K, W)]): DStream[(K, (Iterable[V], Iterable[W]))]  两DStream分别为(K,V)和(K,W)对，返回(K,(Seq[V],Seq[W])对新DStream

        12、def transform[U: ClassTag](transformFunc: RDD[T] => RDD[U]): DStream[U]  将RDD到RDD映射的函数func作用于源DStream中每个RDD上得到新DStream。这个可用于在DStream的RDD上做任意操作。注意的是，在这个转换函数里面能够应用所有RDD的转换操作。

 

3、有状态装换

 

    1、def updateStateByKey[S: ClassTag]( updateFunc: (Seq[V], Option[S]) => Option[S] ): DStream[(K, S)]

           1、 S是你需要保存的状态的类型。

           2、updateFunc 是定义了每一批次RDD如何来更新的状态值。 Seq[V] 是当前批次相同key的值的集合。 Option[S] 是框架自动提供的，上一次保存的状态的值。

           3、updateStateByKey会返回一个新的DStream，该DStream中保存了（Key,State）的序列。

    2、window 函数

           1、def window(windowDuration: Duration, slideDuration: Duration): DStream[T] 基于对源DStream窗化的批次进行计算返回一个新的DStream，windowDuration是窗口大小，slideDuration滑动步长。

           2、def countByWindow( windowDuration: Duration, slideDuration: Duration): DStream[Long]  注意，返回的是window中记录的条数。

           3、def reduceByWindow( reduceFunc: (T, T) => T, windowDuration: Duration, slideDuration: Duration): DStream[T]   通过使用自定义函数整合滑动区间流元素来创建一个新的单元素流。

          4、 def reduceByKeyAndWindow(reduceFunc: (V, V) => V,windowDuration: Duration， slideDuration: Duration

): DStream[(K, V)]   通过给定的窗口大小以滑动步长来应用reduceFunc函数，返回DStream[(K, V)], K就是DStream中相应的K，V是window应用了reduce之后产生的最终值。

           5、def reduceByKeyAndWindow(reduceFunc: (V, V) => V,invReduceFunc: (V, V) => V,windowDuration: Duration,slideDuration: Duration = self.slideDuration,

numPartitions: Int = ssc.sc.defaultParallelism,

filterFunc: ((K, V)) => Boolean = null

): DStream[(K, V)]

           这个版本是4版本的优化版本，计算方式不同，采用增量计算的模式，每次计算只是在上一次计算的基础上减去丢失的值，加上增加的值。