大家好，我是不温卜火，是一名计算机学院大数据专业大二的学生，昵称来源于成语—不温不火，本意是希望自己性情温和。作为一名互联网行业的小白，博主写博客一方面是为了记录自己的学习过程，另一方面是总结自己所犯的错误希望能够帮助到很多和自己一样处于起步阶段的萌新。但由于水平有限，博客中难免会有一些错误出现，有纰漏之处恳请各位大佬不吝赐教！暂时只有csdn这一个平台，博客主页：https://buwenbuhuo.blog.csdn.net/

  本片博文为大家带来的是DStream WorldCount入门案例与解析。

  一说到入门案例，我们就不得不提wordcount这个案例了哈哈哈。所以 此次还是这个案例

一. 案例

1. 需求

  使用 netcat 工具向 9999 端口不断的发送数据，通过 Spark Streaming 读取端口数据并统计不同单词出现的次数

2. 添加依赖

<dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming_2.11</artifactId> <version>2.1.1</version>
</dependency>


  

 

  
1
  
2
  
3
  
4
  
5
  
6
 

3. 源码

package com.buwenbuhuo.spark.streaming.day01

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @author 不温卜火
 * @create 2020-08-07 11:41 MyCSDN ：  https://buwenbuhuo.blog.csdn.net/
 */
object WordCount {
  def main(args: Array[String]): Unit = { // 1. 创建StreamingContext val conf: SparkConf = new SparkConf().setMaster("local[2]").setAppName("WordCount") val ssc: StreamingContext = new StreamingContext(conf,Seconds(3)) // 2. 从数据源创建一个流： socket，rdd队列，自定义接收器 ，kafka(重点) val sourceStream: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop002",9999) // 3. 对流做各种转换 val result: DStream[(String, Int)] = sourceStream.flatMap(_.split(" ")).map((_,1)).reduceByKey(_ + _) // 4. 行动算子 print foreach foreachRDD result.print()  //  把结果打印在控制台 // 5. 启动流 ssc.start() // 6. 阻止主线程退出(阻塞主线程) ssc.awaitTermination() }
}


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
  
20
  
21
  
22
  
23
  
24
  
25
  
26
  
27
  
28
  
29
  
30
  
31
  
32
  
33
  
34
  
35
  
36
  
37
 

4 .测试

• 1. hadoop002上启动 netcat

• 2. 可以打包到 linux 启动我们的 wordcount, 也可以在 idea 直接启动.
• 3. 查看输出结果. 每 3 秒统计一次数据的输入情况.

5. 注意

1. 一旦StreamingContext已经启动, 则不能再添加添加新的 streaming computations

2. 一旦一个StreamingContext已经停止(StreamingContext.stop()), 他也不能再重启

3. 在一个 JVM 内, 同一时间只能启动一个StreamingContext

4. stop() 的方式停止StreamingContext, 也会把SparkContext停掉. 如果仅仅想停止StreamingContext, 则应该这样: stop(false)

5. 一个SparkContext可以重用去创建多个StreamingContext, 前提是以前的StreamingContext已经停掉,并且SparkContext没有被停掉

二. 案例解析

  Discretized Stream(DStream) 是 Spark Streaming 提供的基本抽象, 表示持续性的数据流, 可以来自输入数据, 也可以是其他的 DStream 转换得到. 在内部, 一个 DSteam 用连续的一系列的 RDD 来表示. 在 DStream 中的每个 RDD 包含一个确定时间段的数据.

  对 DStream 的任何操作都会转换成对他里面的 RDD 的操作. 比如前面的 wordcount 案例, flatMap是应用在 line DStream 的每个 RDD 上, 然后生成了 words SStream 中的 RDD. 如下图所示:

  对这些 RDD 的转换是有 Spark 引擎来计算的. DStream 的操作隐藏的大多数的细节, 然后给开发者提供了方便使用的高级 API.

  本次的分享就到这里了,

  好书不厌读百回，熟读课思子自知。而我想要成为全场最靓的仔，就必须坚持通过学习来获取更多知识，用知识改变命运，用博客见证成长，用行动证明我在努力。
  如果我的博客对你有帮助、如果你喜欢我的博客内容，请“点赞” “评论”“收藏”一键三连哦！听说点赞的人运气不会太差，每一天都会元气满满呦！如果实在要白嫖的话，那祝你开心每一天，欢迎常来我博客看看。
  码字不易，大家的支持就是我坚持下去的动力。点赞后不要忘了关注我哦！

文章来源: buwenbuhuo.blog.csdn.net，作者：不温卜火，版权归原作者所有，如需转载，请联系作者。

原文链接：buwenbuhuo.blog.csdn.net/article/details/107856876