二、Spark Streaming DStream操作

太过爱你忘了你带给我的痛 2024-04-17 21:54 72阅读 0赞

# 一、DStream转换操作 #

## 1、DStream无状态转换操作 ##

每次统计，和之前批次无关，不会进行累计

1.  map(func) ：对源DStream的每个元素，采用func函数进行转换，得到一个新的Dstream
2.  flatMap(func)： 与map相似，但是每个输入项可用被映射为0个或者多个输出项
3.  filter(func)： 返回一个新的DStream，仅包含源DStream中满足函数func的项
4.  repartition(numPartitions)： 通过创建更多或者更少的分区改变DStream的并行程度
5.  reduce(func)：利用函数func聚集源DStream中每个RDD的元素，***返回一个包含单元素RDDs的新DStream***
6.  count()：统计源DStream中***每个RDD的元素数量***
7.  union(otherStream)： 返回一个新的DStream，包含源DStream和其他DStream的元素
8.  countByValue()：应用于元素类型为(K,V)的DStream上，返回一个（K，V）键值对类型的新DStream，每个键的值是在原DStream的每个RDD中的出现次数
9.  reduceByKey(func, \[numTasks\])：当在一个由(K,V)键值对组成的DStream上执行该操作时，返回一个新的由(K,V)键值对组成的DStream，每一个key的值均由给定的recuce函数（func）聚集起来
10. join(otherStream, \[numTasks\])：当应用于两个DStream（一个包含（K,V）键值对,一个包含(K,W)键值对），返回一个包含(K, (V, W))键值对的新Dstream
11. cogroup(otherStream, \[numTasks\])：当应用于两个DStream（一个包含（K,V）键值对,一个包含(K,W)键值对），返回一个包含(K, Seq\[V\], Seq\[W\])的元组
12. transform(func)：通过对源DStream的每个RDD应用RDD-to-RDD函数，创建一个新的DStream。支持在新的DStream中做任何RDD操作

## 2、DStream有状态转换操作 ##

### A、滑动窗口转换操作 ###

*  事先设定一个**滑动窗口的长度**（也就是窗口的持续时间）
 *  设定**滑动窗口的时间间隔（每隔多长时间执行一次计算）**，让窗口按照指定时间间隔在源DStream上滑动
 *  每次窗口停放的位置上，都会有一部分Dstream（或者一部分RDD）被框入窗口内，形成一个小段的Dstream
 *  可以启动对这个小段DStream的计算

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70][]

**（1）一些窗口转换操作的含义**

1.  window(windowLength, slideInterval) 基于源DStream产生的窗口化的批数据，计算得到一个新的Dstream
2.  countByWindow(windowLength, slideInterval) 返回**流中元素的一个滑动窗口数**
3.  reduceByWindow(func, windowLength, slideInterval) **返回一个单元素流**。利用函数func聚集滑动时间间隔的流的元素创建这个单元素流。函数func必须满足结合律，从而可以支持并行计算
4.  reduceByKeyAndWindow(func, windowLength, slideInterval, \[numTasks\]) **应用到一个(K,V)键值对组成的DStream上**时，会返回一个由(K,V)键值对组成的新的DStream。每一个key的值均由给定的reduce函数(func函数)进行聚合计算。注意：**在默认情况下，这个算子利用了Spark默认的并发任务数去分组。可以通过numTasks参数的设置来指定不同的任务数**
5.  reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, \[numTasks\]) **更加高效的reduceByKeyAndWindow，**每个窗口的reduce值，是基于先前窗口的reduce值进行增量计算得到的；它会对进入滑动窗口的新数据进行reduce操作，并对离开窗口的老数据进行“逆向reduce”操作。但是，只能用于“可逆reduce函数”，即那些reduce函数都有一个对应的“逆向reduce函数”（以InvFunc参数传入）
6.  countByValueAndWindow(windowLength, slideInterval, \[numTasks\]) 当**应用到一个(K,V)键值对组成的DStream上**，返回一个由(K,V)键值对组成的新的DStream。**每个key的值都是它们在滑动窗口中出现的频率**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 1][]

### B、updateStateByKey操作 ###

需要在**跨批次之间维护状态时**，就必须使用updateStateByKey操作

词频统计实例：

对于有状态转换操作而言，本批次的词频统计，会在之前批次的词频统计结果的基础上进行不断累加，所以，最终统计得到的词频，是所有批次的单词的总的词频统计结果

import org.apache.spark._
    import org.apache.spark.streaming._
    import org.apache.spark.storage.StorageLevel
    object NetworkWordCountStateful {
      def main(args: Array[String]) {
        //定义状态更新函数
        val updateFunc = (values: Seq[Int], state: Option[Int]) => {
          val currentCount = values.foldLeft(0)(_ + _)
          val previousCount = state.getOrElse(0)
          Some(currentCount + previousCount)
        }
        StreamingExamples.setStreamingLogLevels()  //设置log4j日志级别    
        val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCountStateful")
        val sc = new StreamingContext(conf, Seconds(5))
     //设置检查点，检查点具有容错机制
        sc.checkpoint("file:///usr/local/spark/mycode/streaming/stateful/")   
        val lines = sc.socketTextStream("localhost", 9999)
        val words = lines.flatMap(_.split(" "))
        val wordDstream = words.map(x => (x, 1))
        val stateDstream = wordDstream.updateStateByKey[Int](updateFunc)
        stateDstream.print()
        sc.start()
        sc.awaitTermination()
      }
    }

# 二、输出操作 #

在Spark应用中，外部系统经常需要使用到Spark DStream处理后的数据，因此，需要采用输出操作把DStream的数据输出到数据库或者文件系统中

**1 、把DStream输出到文本文件中**

package org.apache.spark.examples.streaming
    import org.apache.spark._
    import org.apache.spark.streaming._
    import org.apache.spark.storage.StorageLevel
    object NetworkWordCountStateful {
      def main(args: Array[String]) {
        //定义状态更新函数
        val updateFunc = (values: Seq[Int], state: Option[Int]) => {
          val currentCount = values.foldLeft(0)(_ + _)
          val previousCount = state.getOrElse(0)
          Some(currentCount + previousCount)
        }
        StreamingExamples.setStreamingLogLevels()  //设置log4j日志级别    
        val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCountStateful")
        val sc = new StreamingContext(conf, Seconds(5))
    //设置检查点，检查点具有容错机制
        sc.checkpoint("file:///usr/local/spark/mycode/streaming/dstreamoutput/")    
        val lines = sc.socketTextStream("localhost", 9999)
        val words = lines.flatMap(_.split(" "))
        val wordDstream = words.map(x => (x, 1))
        val stateDstream = wordDstream.updateStateByKey[Int](updateFunc)
        stateDstream.print()
        //下面是新增的语句，把DStream保存到文本文件中
        stateDstream.saveAsTextFiles("file:///usr/local/spark/mycode/streaming/dstreamoutput/output.txt")
        sc.start()
        sc.awaitTermination()
      }
    }

**2、把DStream写入到MySQL数据库中**

package org.apache.spark.examples.streaming
    import java.sql.{PreparedStatement, Connection, DriverManager}
    import java.util.concurrent.atomic.AtomicInteger
    import org.apache.spark.SparkConf
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    import org.apache.spark.streaming.StreamingContext._
    import org.apache.spark.storage.StorageLevel
    object NetworkWordCountStateful {
      def main(args: Array[String]) {
        //定义状态更新函数
        val updateFunc = (values: Seq[Int], state: Option[Int]) => {
          val currentCount = values.foldLeft(0)(_ + _)
          val previousCount = state.getOrElse(0)
          Some(currentCount + previousCount)
        }
        StreamingExamples.setStreamingLogLevels()  //设置log4j日志级别
    
        val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCountStateful")
        val sc = new StreamingContext(conf, Seconds(5))
    //设置检查点，检查点具有容错机制
        sc.checkpoint("file:///usr/local/spark/mycode/streaming/dstreamoutput/")    
        val lines = sc.socketTextStream("localhost", 9999)
        val words = lines.flatMap(_.split(" "))
        val wordDstream = words.map(x => (x, 1))
        val stateDstream = wordDstream.updateStateByKey[Int](updateFunc)
        stateDstream.print()
        //下面是新增的语句，把DStream保存到MySQL数据库中
        stateDstream.foreachRDD(rdd => {
          //内部函数
          def func(records: Iterator[(String,Int)]) {
            var conn: Connection = null
            var stmt: PreparedStatement = null
            try {
              val url = "jdbc:mysql://localhost:3306/spark"
              val user = "root"
              val password = "hadoop"  //数据库密码是hadoop
              conn = DriverManager.getConnection(url, user, password)
              records.foreach(p => {
                val sql = "insert into wordcount(word,count) values (?,?)"
                stmt = conn.prepareStatement(sql);
                stmt.setString(1, p._1.trim)
                stmt.setInt(2,p._2.toInt)
                stmt.executeUpdate()
              })
            } catch {
              case e: Exception => e.printStackTrace()
            } finally {
              if (stmt != null) {
                stmt.close()
              }
              if (conn != null) {
                conn.close()
              }
            }
          }
          val repartitionedRDD = rdd.repartition(3)
          repartitionedRDD.foreachPartition(func)
        })
        sc.start()
        sc.awaitTermination()
      }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/bfff524771974706965a6439aee889d3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/9bf4edae986f4ff19b1885892e5e6bd1.png

发表评论取消回复

表情：

评论列表（有 0 条评论，72人围观）

还没有评论，来说两句吧...

相关阅读

相关 Spark Dstream

2.Dstream 入门 **Spark Streaming的入口** **StreamingContext** val conf = new Spark...

浅浅的花香味﹌/ 2024年04月20日 09:10/ 0 赞/ 64 阅读

相关二、Spark Streaming DStream操作

一、DStream转换操作 1、DStream无状态转换操作每次统计，和之前批次无关，不会进行累计 1. map(func) ：对源DStream的每个元...

太过爱你忘了你带给我的痛/ 2024年04月17日 21:54/ 0 赞/ 73 阅读

相关 [Spark、hadoop]spark Streaming的核心DStream

目录 Spark Streaming的核心是DStream 一、DStream简介二．DStream编程模型三．DStream转换操作 -------------

Bertha 。/ 2024年04月03日 08:48/ 0 赞/ 91 阅读

相关 SparkStreaming（二）：DStream操作

文章目录一、DStream创建 1.1 RDD 队列 1.2 自定义数据源 1.3 Kafka 数据源

Bertha 。/ 2022年12月24日 09:54/ 0 赞/ 401 阅读

相关 Apache Spark 3.0 DStreams-Streaming编程指南

目录总览一个简单的例子基本概念连结中初始化StreamingContext 离散流（DStreams）输入DStreams和接收器基本资料进阶

超、凢脫俗/ 2022年12月04日 07:50/ 0 赞/ 227 阅读

相关 Spark Streaming——DStream Transformation操作

Spark Streaming——DStream Transformation操作 Spark的各个子框架都是基于spark core的，Spark Streaming在

r囧r小猫/ 2022年06月08日 08:15/ 0 赞/ 270 阅读

相关 Spark系列--Spark Streaming(七)DStreams输出

前言输出操作指定了对流数据经转化操作得到的数据所要执行的操作(例如把结果推入外部数据库或输出到屏幕上)。与 RDD 中的惰性求值类似，如果一个 DStream 及其派生

青旅半醒/ 2022年05月18日 08:12/ 0 赞/ 261 阅读

相关 Spark系列--Spark Streaming(六)DStreams转换

前言 DStream上的原语与RDD的类似，分为Transformations（转换）和Output Operations（输出）两种，此外转换操作中还有一些比较特殊的原

た入场券/ 2022年05月18日 07:40/ 0 赞/ 240 阅读

相关 [spark streaming] DStream 和 DStreamGraph 解析

前言 Spark Streaming 是基于Spark Core将流式计算分解成一系列的小批处理任务来执行。在Spark Streaming里，总体负责任务的动态调度

冷不防/ 2022年05月15日 10:26/ 0 赞/ 190 阅读

相关 Spark之SparkStreaming的DStream操作

DStream的转换操作和输出、累加器等：转换 DStream上的原语分为Transformations（转换）和Output Operations（输出）两种，此

Dear 丶/ 2022年01月21日 07:29/ 0 赞/ 290 阅读