Spark：Streaming 实践 Dstream 转换算子、窗口、输出文件

转x33g5p2x  于2021-12-07 转载在 Spark  
字(11.6k)|赞(0)|评价(0)|浏览(770)

1、启动集群

  1. 启动zookeeper,hadoop,flume
  2. # 1、三个节点
  3. /usr/zookeeper/zookeeper-3.4.10/bin/zkServer.sh start 
  4. /usr/zookeeper/zookeeper-3.4.10/bin/zkServer.sh status
  6. # 2、master节点 启动hadoop
  7. /usr/hadoop/hadoop-2.7.3/sbin/start-all.sh

2、IDEA安装依赖

  1. <properties>
  2.         <scala.version>2.11.8</scala.version>
  3.         <hadoop.version>2.7.3</hadoop.version>
  4.         <spark.version>2.4.0</spark.version>
  5.         <hbase.version>1.2.4</hbase.version>
  6.         <hive.version>2.1.1</hive.version>
  7.     </properties>
  9.     <dependencies>
  10.         <!--Scala-->
  11.         <dependency>
  12.             <groupId>org.scala-lang</groupId>
  13.             <artifactId>scala-library</artifactId>
  14.             <version>${scala.version}</version>
  15.         </dependency>
  17.         <!--Spark-->
  18.         <dependency>
  19.             <groupId>org.apache.spark</groupId>
  20.             <artifactId>spark-core_2.11</artifactId>
  21.             <version>${spark.version}</version>
  22.         </dependency>
  24.         <dependency>
  25.             <groupId>org.apache.spark</groupId>
  26.             <artifactId>spark-streaming_2.11</artifactId>
  27.             <version>2.4.0</version>
  28.         </dependency>
  30.         <dependency>
  31.             <groupId>org.apache.spark</groupId>
  32.             <artifactId>spark-sql_2.11</artifactId>
  33.             <version>2.4.0</version>
  34.         </dependency>
  36.         <dependency>
  37.             <groupId>org.apache.spark</groupId>
  38.             <artifactId>spark-mllib_2.11</artifactId>
  39.             <version>${spark.version}</version>
  40.         </dependency>
  42.         <dependency>
  43.             <groupId>org.apache.spark</groupId>
  44.             <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
  45.             <version>${spark.version}</version>
  46.         </dependency>
  48.         <!--Spark & flume-->
  49.         <dependency>
  50.             <groupId>org.apache.spark</groupId>
  51.             <artifactId>spark-streaming-flume_2.11</artifactId>
  52.             <version>2.3.0</version>
  53.         </dependency>
  55.         <!--Hadoop-->
  56.         <dependency>
  57.             <groupId>org.apache.hadoop</groupId>
  58.             <artifactId>hadoop-client</artifactId>
  59.             <version>${hadoop.version}</version>
  60.         </dependency>
  62.         <dependency>
  63.             <groupId>org.apache.hadoop</groupId>
  64.             <artifactId>hadoop-common</artifactId>
  65.             <version>2.7.3</version>
  66.         </dependency>
  68.         <dependency>
  69.             <groupId>org.apache.hadoop</groupId>
  70.             <artifactId>hadoop-hdfs</artifactId>
  71.             <version>2.7.3</version>
  72.         </dependency>
  74.         <!--Hbase-->
  75.         <dependency>
  76.             <groupId>org.apache.hbase</groupId>
  77.             <artifactId>hbase-client</artifactId>
  78.             <version>${hbase.version}</version>
  79.         </dependency>
  80.         <dependency>
  81.             <groupId>org.apache.hbase</groupId>
  82.             <artifactId>hbase-common</artifactId>
  83.             <version>${hbase.version}</version>
  84.         </dependency>
  86.         <dependency>
  87.             <groupId>org.apache.hbase</groupId>
  88.             <artifactId>hbase-server</artifactId>
  89.             <version>${hbase.version}</version>
  90.         </dependency>
  92.         <dependency>
  93.             <groupId>org.apache.hbase</groupId>
  94.             <artifactId>hbase-protocol</artifactId>
  95.             <version>${hbase.version}</version>
  96.         </dependency>
  98.         <dependency>
  99.             <groupId>org.apache.hbase</groupId>
  100.             <artifactId>hbase-annotations</artifactId>
  101.             <version>${hbase.version}</version>
  102.             <type>test-jar</type>
  103.             <scope>test</scope>
  104.         </dependency>
  106.         <!--Hive-->
  107.         <dependency>
  108.             <groupId>org.apache.hive</groupId>
  109.             <artifactId>hive-exec</artifactId>
  110.             <version>${hive.version}</version>
  111.         </dependency>
  112.         <dependency>
  113.             <groupId>org.apache.hive</groupId>
  114.             <artifactId>hive-jdbc</artifactId>
  115.             <version>${hive.version}</version>
  116.         </dependency>
  118.         <!--kafka-->
  119.         <dependency>
  120.             <groupId>org.apache.kafka</groupId>
  121.             <artifactId>kafka-clients</artifactId>
  122.             <version>2.4.0</version>
  123.         </dependency>
  125.         <dependency>
  126.             <groupId>org.apache.kafka</groupId>
  127.             <artifactId>kafka-streams</artifactId>
  128.             <version>2.4.0</version>
  129.         </dependency>
  131.         <!--mysql-->
  132.         <dependency>
  133.             <groupId>mysql</groupId>
  134.             <artifactId>mysql-connector-java</artifactId>
  135.             <version>5.1.46</version>
  136.         </dependency>
  138.         <dependency>
  139.             <groupId>com.alibaba</groupId>
  140.             <artifactId>fastjson</artifactId>
  141.             <version>1.2.17</version>
  142.         </dependency>
  144.         <dependency>
  145.             <groupId>com.huaban</groupId>
  146.             <artifactId>jieba-analysis</artifactId>
  147.             <version>1.0.2</version>
  148.         </dependency>
  150.         <dependency>
  151.             <groupId>junit</groupId>
  152.             <artifactId>junit</artifactId>
  153.             <version>4.12</version>
  154.             <scope>test</scope>
  155.         </dependency>
  156.     </dependencies>
  158.     <build>
  159.         <plugins>
  160.             <plugin>
  161.                 <groupId>org.scala-tools</groupId>
  162.                 <artifactId>maven-scala-plugin</artifactId>
  163.                 <version>2.15.2</version>
  164.                 <executions>
  165.                     <execution>
  166.                         <goals>
  167.                             <goal>compile</goal>
  168.                             <goal>testCompile</goal>
  169.                         </goals>
  170.                     </execution>
  171.                 </executions>
  172.             </plugin>
  174.             <plugin>
  175.                 <groupId>org.apache.maven.plugins</groupId>
  176.                 <artifactId>maven-compiler-plugin</artifactId>
  177.                 <version>3.1</version>
  178.                 <configuration>
  179.                     <source>1.8</source>
  180.                     <target>1.8</target>
  181.                 </configuration>
  182.             </plugin>
  184.             <plugin>
  185.                 <groupId>org.apache.maven.plugins</groupId>
  186.                 <artifactId>maven-surefire-plugin</artifactId>
  187.                 <version>2.12.4</version>
  188.                 <configuration>
  189.                     <skip>true</skip>
  190.                 </configuration>
  191.             </plugin>
  192.         </plugins>
  193.         <defaultGoal>compile</defaultGoal>
  194.     </build>
  195. </project>

在之前安装过的只需要安装：

  1. <dependency>
  2.             <groupId>org.apache.kafka</groupId>
  3.             <artifactId>kafka-streams</artifactId>
  4.             <version>2.4.0</version>
  5.         </dependency>

2.1 启动服务端监听 Socket 服务

命令：nc -lk 9999

2.2 实现 transform() 方法，分割多个单词

  1. import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
  2. import org.apache.spark.streaming.{Seconds, StreamingContext}
  3. import org.apache.spark.{SparkConf, SparkContext}
  5. object TransformTest {
  6.   def main(args: Array[String]): Unit = {
  7.     // 1.创建SparkConf对象
  8.     val sparkConf: SparkConf = new SparkConf()
  9.       .setAppName("TransformTest").setMaster("local[2]")
  10.     // 2.创建SparkContext对象，它是所有任务计算的源头
  11.     val sc: SparkContext = new SparkContext(sparkConf)
  12.     // 3.设置日志级别
  13.     sc.setLogLevel("WARN")
  14.     // 4.创建StreamingContext，需要两个参数，分别为SparkContext和批处理时间间隔
  15.     val ssc: StreamingContext = new StreamingContext(sc,Seconds(5))
  16.     // 5.连接socket服务，需要socket服务地址、端口号及存储级别(默认的)
  17.     // 以上是固定搭配结构
  18.     val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.142.128",9999)
  19.     
  20.     // 6.使用RDD-to-RDD函数，返回新的DStream对象（即words），并空格切分每行
  21.     val words: DStream[String] = dstream.transform(rdd => rdd.flatMap(_.split(" ")))
  22.     // 7.打印输出结果
  23.     words.print()
  24.     // 8.开启流式计算
  25.     ssc.start()
  26.     // 9.让程序一直运行，除非人为干预停止
  27.     ssc.awaitTermination()
  28.   }
  29. }

2.3 UpdateStateByKeyTest 更新值

  1. import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
  2. import org.apache.spark.streaming.{Seconds, StreamingContext}
  3. import org.apache.spark.{SparkConf, SparkContext}
  4. object UpdateStateByKeyTest {
  5.   //newValues 表示当前批次汇总成的(word,1)中相同单词的所有1
  6.   //runningCount 表示历史的所有相同key的value总和
  7.   def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
  8.     val newCount =runningCount.getOrElse(0)+newValues.sum
  9.     Some(newCount)
  10.   }
  11.   def main(args: Array[String]): Unit = {
  12.     // 1.创建SparkConf对象 设置appName和master地址 local[2] 表示本地采用2个线程运行任务
  13.     val sparkConf: SparkConf = new SparkConf().setAppName("WordCount").setMaster("local[2]")
  14.     // 2.创建SparkContext对象,它是所有任务计算的源头，它会创建DAGScheduler和TaskScheduler
  15.     val sc: SparkContext = new SparkContext(sparkConf)
  16.     // 3.设置日志级别
  17.     sc.setLogLevel("WARN")
  18.     // 4.创建StreamingContext,需要2个参数，一个是SparkContext，一个是批处理的时间间隔
  19.     val ssc: StreamingContext = new StreamingContext(sc,Seconds(2))
  20.     
  21.     // 5.配置检查点目录，使用updateStateByKey方法必须配置检查点目录
  22.     ssc.checkpoint("./")
  23.     // 6.对接socket数据创建DStream对象，需要socket服务的地址、端口号及存储级别(默认的)
  24.     val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("master",9999)
  25.     
  26.     // 7.按空格进行切分每一行,并将切分的单词出现次数记录为1
  27.     val wordAndOne: DStream[(String, Int)] = dstream.flatMap(_.split(" ")).map(word =>(word,1))
  28.     // 8.调用updateStateByKey操作，统计单词在全局中出现的次数
  29.     var result: DStream[(String, Int)] = wordAndOne.updateStateByKey(updateFunction)
  30.     // 9.打印输出结果
  31.     result.print()
  32.     // 10.开启流式计算
  33.     ssc.start()
  34.     // 11.让程序一直运行，除非人为干预停止
  35.     ssc.awaitTermination()
  36.   }
  37. }

2.4 Dstream 窗口操作

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

方法名称相关说明
window(windowLength, slideInterval)基于源DStream产生的窗口化的批数据，计算得到一个新的Dstream；
countByWindow(windowLength, slideInterval)返回流中元素的一个滑动窗口数；
reduceByWindow(func, windowLength, slideInterval)返回一个单元素流。利用函数func聚集滑动时间间隔的流的元素创建这个单元素流。函数func必须满足结合律，从而可以支持并行计算；
reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks])应用到一个(K,V)键值对组成的DStream上时，会返回一个由(K,V)键值对组成的新的DStream。每一个key的值均由给定的reduce函数(func函数)进行聚合计算。注意：在默认情况下，这个算子利用了Spark默认的并发任务数去分组。可以通过numTasks参数的设置来指定不同的任务数。
reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks])更加高效的 reduceByKeyAndWindow，每个窗口的reduce值，是基于先前窗口的reduce值进行增量计算得到的；它会对进入滑动窗口的新数据进行reduce操作，并对离开窗口的老数据进行“逆向reduce‖操作。但是，只能用于“可逆reduce函数”，即那些reduce函数都有一个对应的“逆向reduce函数”（以InvFunc参数传入）。
countByValueAndWindow(windowLength, slideInterval, [numTasks])当应用到一个(K,V)键值对组成的DStream上，返回一个由(K,V)键值对组成的新的DStream；每个key的值都是它们在滑动窗口中出现的频率。
  1. import org.apache.spark.{SparkConf, SparkContext}
  2. import org.apache.spark.streaming.{Seconds, StreamingContext}
  3. import org.apache.spark.streaming.dstream.{DStream,ReceiverInputDStream}
  4. object WindowTest {
  5.   def main(args: Array[String]): Unit = {
  6.     // 1.创建SparkConf对象
  7.     val sparkConf: SparkConf = new SparkConf()
  8.       .setAppName("WindowTest ").setMaster("local[2]")
  9.     // 2.创建SparkContext对象，它是所有任务计算的源头
  10.     val sc: SparkContext = new SparkContext(sparkConf)
  11.     // 3.设置日志级别
  12.     sc.setLogLevel("WARN")
  13.     // 4.创建StreamingContext，需要两个参数，分别为SparkContext和批处理时间间隔
  14.     val ssc: StreamingContext = new StreamingContext(sc,Seconds(1))
  15.     // 5.连接socket服务，需要socket服务地址、端口号及存储级别(默认的)
  16.     val dstream: ReceiverInputDStream[String] = ssc
  17.       .socketTextStream("master",9999)
  18.       
  19.     // 6.按空格进行切分每一行
  20.     val words: DStream[String] = dstream.flatMap(_.split(" "))
  21.     // 7.调用window操作，需要两个参数，窗口长度和滑动时间间隔
  22.     val windowWords: DStream[String] = words.window(Seconds(3),Seconds(1))
  23.     // 8.打印输出结果
  24.     windowWords.print()
  25.     // 9.开启流式计算
  26.     ssc.start()
  27.     // 10.让程序一直运行，除非人为干预停止
  28.     ssc.awaitTermination()
  29.   }
  30. }

把批处理时间间隔、窗口长度和滑动时间间隔进行变换，在其中两者非整数倍的情况下，会报错。

2.5 DStream 输出操作

  1. import org.apache.spark.{SparkConf, SparkContext}
  2. import org.apache.spark.streaming.{Seconds, StreamingContext}
  3. import org.apache.spark.streaming.dstream.ReceiverInputDStream
  5. object SaveAsTextFilesTest {
  6.   def main(args: Array[String]): Unit = {
  7.     System.setProperty("HADOOP_USER_NAME", "root")
  8.     //1.创建SparkConf对象 设置appName和master地址 local[2] 表示本地采用2个线程运行任务
  9.     val sparkConf: SparkConf = new SparkConf().setAppName("SaveAsTextFilesTest").setMaster("local[2]")
  10.     //2.创建SparkContext对象,它是所有任务计算的源头，它会创建DAGScheduler和TaskScheduler
  11.     val sc: SparkContext = new SparkContext(sparkConf)
  12.     //3.设置日志级别
  13.     sc.setLogLevel("WARN")
  14.     //4.创建StreamingContext,需要2个参数，一个是SparkContext，一个是批处理的时间间隔
  15.     val ssc: StreamingContext = new StreamingContext(sc,Seconds(5))
  16.     //5.对接socket数据创建DStream对象，需要socket服务的地址、端口号及存储级别(默认的)
  17.     val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.142.128",9999)
  18.     
  19.     //6.调用saveAsTextFiles操作，将nc窗口输出的内容保存到HDFS上
  20.     dstream.saveAsTextFiles("hdfs://master:8020//saveAsTextFiles/satf","txt")
  21.     //7.开启流式计算
  22.     ssc.start()
  23.     //8.让程序一直运行，除非人为干预停止
  24.     ssc.awaitTermination()
  25.   }
  26. }

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://blog.csdn.net/weixin_44775255/article/details/121765631

内容来源于网络，如有侵权，请联系作者删除！

sparkflumestreaming大数据大数据 linux

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com