【大数据技术Spark】DStream编程操作讲解实战（图文解释附源码）

女爷i 2023-09-30 18:08 37阅读 0赞

## DStream编程 ##

批处理引擎Spark Core把输入的数据按照一定的时间片（如1s）分成一段一段的数据，每一段数据都会转换成RDD输入到Spark Core中，然后将DStream操作转换为RDD算子的相关操作，即转换操作、窗口操作以及输出操作。RDD算子操作产生的中间结果数据会保存在内存中，也可以将中间的结果数据输出到外部存储系统中进行保存。

## 转换操作 ##

### １：无状态转换操作 ###

无状态转化操作每个批次的处理不依赖于之前批次的数据。常见的 RDD 转化操作，例如 Map()、filter()、ReduceByKey() 等，都是无状态转化操作。

![eefd25f22c6444949767d777f1df875d.png][]

![9949a87ce0b048c689bf0b8d865ebf78.png][]

![b5f0fc6d7af94030892ce92df048683e.png][]

### ２：有状态转化操作： ###

有状态转化操作需要使用之前批次的数据或者是中间结果来计算当前批次的数据。有状态转化操作包括基于滑动窗口的转化操作和追踪状态变化的转化操作。

![66c8d8a3e43445b2ac1b58b530c6e6f7.png][]

DStream API提供的与窗口操作相关的方法

![f439f5cc18dc4395a029feaea4b29229.png][]

DStream API提供的与输出操作相关的方法

![f939e3e95fba4a01a22f29f2fb00fb9b.png][]

编写Spark Streaming程序的基本步骤是：

1）通过创建输入DStream来定义输入源。

2）通过对DStream应用转换操作和输出操作来定义流计算。

3）用streamingContext.start()来开始接收数据和处理流程

4）通过streamingContext.awaitTermination()方法来等待处理结束（手动结束或因为错误而结束）。

5）可以通过streamingContext.stop()来手动结束流计算进程。

通过示例进行演示，此示例为监视一个文件夹的log日志，并计算每个单词出现的次数

cogroup和join算子需要两个并行数据流，对两个数据流直接关联，不同的是join算子是把两个RDD按照相同的key拼在一起，类似SQL中的等值连接，可以类似的使用其他算子进行RDD的左连接等，而cogroup算子是把两个RDD按照key拼起来，但是它会汇总得到的value，最后的结果的条数是根据key决定的，有多少key就汇总成多少条数据，然后把RDD的所有相同的key的value放到一个Iterable里面，类似于SQL里面的全连接

设置为本地运行模式，2个线程，一个监听，另一个处理数据
        val sparkConf = new SparkConf().setAppName("WordCountStreaming").setMaster("local[2]")
        // 时间间隔为20秒
        val stc = new StreamingContext(sparkConf， Seconds(20))
        //定义输入源，监听本地目录，也可以采用HDFS文件
        val lines = stc.textFileStream("E:/log")
        //应用转换操作flatMap流计算
        val words = lines.flatMap(_.split(" "))
        //应用转换操作Map和ReduceByKey计算
        val wordCounts = words.Map(x => (x， 1)).ReduceByKey(_ + _)
        wordCounts.print()
        //开始接收数据和处理流程
        stc.start()
        //等待处理结束
        stc.awaitTermination()
    //创建两个可被并行操作的分布式数据集
        val idName = sc.parallelize(Array((1， "张三")， (2， "李四")， (3， "王五")))
        val idAge = sc.parallelize(Array((1， 30)， (2， 29)， (4， 21)))
        println("\ncogroup\n")
        //对两个并行数据集进行cogroup操作
        idName.cogroup(idAge).collect().foreach(println)
        println("\njoin\n")
        //对两个并行数据集进行join操作
         idName.join(idAge).collect().foreach(println)
     //3.获取StreamingContext对象，5秒一个批次
        val ssc = new StreamingContext(sparkContext，Seconds(5))
        //4.接收socket的数据
        val textStream: ReceiverInputDStream[String] = ssc.socketTextStream("hhaonote"，9999)
        //5.获取每一行的单词
        val words: DStream[String] = textStream.flatMap(_.split(" "))
        //6.为每一个单词置为1
        val wordAndOne: DStream[(String， Int)] = words.Map((_，1))
         //7.每隔10秒统计最近10秒的搜索词出现的次数
        val result: DStream[(String， Int)] = wordAndOne.ReduceByKeyAndWindow((x:Int，y:Int)=>x+y，Seconds(10)，Seconds(10))
        //8.打印
        result.print()

**创作不易 觉得有帮助请点赞关注收藏~~~**

[eefd25f22c6444949767d777f1df875d.png]: https://img-blog.csdnimg.cn/eefd25f22c6444949767d777f1df875d.png
[9949a87ce0b048c689bf0b8d865ebf78.png]: https://img-blog.csdnimg.cn/9949a87ce0b048c689bf0b8d865ebf78.png
[b5f0fc6d7af94030892ce92df048683e.png]: https://img-blog.csdnimg.cn/b5f0fc6d7af94030892ce92df048683e.png
[66c8d8a3e43445b2ac1b58b530c6e6f7.png]: https://img-blog.csdnimg.cn/66c8d8a3e43445b2ac1b58b530c6e6f7.png
[f439f5cc18dc4395a029feaea4b29229.png]: https://img-blog.csdnimg.cn/f439f5cc18dc4395a029feaea4b29229.png
[f939e3e95fba4a01a22f29f2fb00fb9b.png]: https://img-blog.csdnimg.cn/f939e3e95fba4a01a22f29f2fb00fb9b.png