- 一、实验目的
- 二、实验内容
- 三、实验原理
- 四、实验环境
- 五、实验步骤
- 5.1 启动Spark集群
- 5.2 使用updateStateByKey函数，每5秒钟计算一次每个单词出现的累加数量
- 5.3 使用mapWithState函数，每5秒钟计算一次每个单词出现的累加数量
申明： 未经许可，禁止以任何形式转载，若要引用，请标注链接地址
全文共计3695字，阅读大概需要3分钟 一、实验目的
掌握 DStream数据累加函数updateStateByKey。
掌握 DStream数据累加函数mapWithState。
1、每5秒钟计算一次每个单词出现的累加数量。（使用Socket数据源）
三、实验原理在DStream中支持跨批次数据执行计算时保持任意状态。在Spark第一代流处理(Spark Streaming)中，这需要手工实现。
四、实验环境硬件：x86_64 ubuntu 16.04服务器
软件：JDK 1.8，Spark-2.3.2，Hadoop-2.7.3
1、在终端窗口下，输入如下命令，启动Spark集群：
1. $ start-dfs.sh 2. $ cd /opt/spark 3. $ ./sbin/start-all.sh
2、启动spark-shell。在终端窗口下,输入如下命令（注意：以下命令中的localhost，请换成你自己虚拟机实际的机器名）：
1. $ spark-shell --master spark://localhost:7077
3、另打开一个终端窗口，键入以下命令，启动Socket服务器：
1. $ nc -lp 99995.2 使用updateStateByKey函数，每5秒钟计算一次每个单词出现的累加数量
1、updateStateByKey 解释:
以DStream中的数据进行按key做reduce操作，然后对各个批次的数据进行累加。在有新的数据信息进入或更新时。能够让用户保持想要的不论什么状。使用这个函数需要两步：
1) 定义状态：可以是随意数据类型；
2) 定义状态更新函数：用一个函数指定怎样使用先前的状态。从输入流中的新值更新状态。
下面的实验代码使用updateStateByKey(newUpdateFunc)函数计算每5秒钟单词出现的累加数量。
请在spark-shell的paste模式下，输入以下代码：
1. import org.apache.spark.SparkContext
2. import org.apache.spark.sql.catalyst.expressions.Second
3. import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
4. import org.apache.spark.streaming.dstream.DStream
5. import org.apache.spark.streaming.{Seconds, StreamingContext}
6.
7. val ssc = new StreamingContext(sc, Seconds(5))
8.
9. // 设置检查点缓存目录
10. ssc.checkpoint("hdfs://localhost:9000/checkpoints")
11.
12. // 监听nc服务端口
13. val lines = ssc.socketTextStream("localhost", 9999)
14.
15. // 把每行数据安“空格”分隔
16. val words = lines.flatMap(_.split(" "))
17.
18. // 把单和1组合
19. val word_count = words.map((_, 1))
20.
21. // 定义一个数据处理函数
22. val updateFunc = (values: Seq[Int], state: Option[Int]) => {
23. // values 是所有相同key的value集
24. // state 是旧数据状态值 当第一没有值是返回0
25. val count = values.sum + state.getOrElse(0)
26. Some(count)
27. }
28.
29. // updateStateByKey对数据进行 累加
30. val words_count = word_count.updateStateByKey(updateFunc)
31.
32. // 打印数据
33. words_count.print()
34.
35. ssc.start()
36. ssc.awaitTermination()
2、切换到另一个终端窗口（Socket服务器所运行的窗口），随意输入一些语句，语句间的单词以空格分隔。如下所示：
1. java python 2. ... 3. java python scala
请确保输入的语句跨5秒钟。
3、切换到spark-shell所在终端窗口，查看打印出的结果，结果如下所示：
—————————————————————- Time: 1548654660000 ms —————————————————————- —————————————————————- Time: 1548654665000 ms —————————————————————- (python,1) (java,1) —————————————————————- Time: 1548654670000 ms —————————————————————- (scala,1) (python,2) (java,2)
由上面的内容可以看出，第一次和第二次的数据进行了累加统计。
4、停止Spark流计算。在spark-shell窗口中，执行以下命令，停止流计算的执行：
1. ssc.stop(false)
或者，同时按下【Ctrl + C】键来终止。
5.3 使用mapWithState函数，每5秒钟计算一次每个单词出现的累加数量1、mapWithStat 和 updateStateByKey一样都是状态管理，返回一个MapWithStateDStream。它使用一个函数不断作用于这个DStream中key- value的元素，基于key进行状态维护和更新。mapWithState这个方法参数是StateSpec，这里边封装了对数据操作的函数。
下面使用mapWithState函数，同样每5秒钟计算一次每个单词出现的累加数量。
在spark-shell窗口中，在paste式下，输入以下代码:
1. import org.apache.spark.streaming.State
2. import org.apache.spark.streaming.StateSpec
3.
4. // 定义一个函数
5. val mappingFunc = (word: String, count: Option[Int], state: State[Int]) => {
6. // count 获取数据的value值, 与state的状态值相加
7. val sum = count.getOrElse(0) + state.getOption.getOrElse(0)
8.
9. // 输出累加后的统计结果
10. val output = (word, sum)
11.
12. // 更新状态值
13. state.update(sum)
14.
15. output
16. }
17.
18. // 定义一个 StateSpec 传入 mappingFunc函数
19. val spc = StateSpec.function(mappingFunc)
20.
21. // 代用mapWithState 传入 spc
22. val value = word_count.mapWithState(spc)
23.
24. // 打印输出
25. value.print()
26.
27. ssc.start()
28. ssc.awaitTermination()
2、切换到另一个终端窗口（Socket服务器所运行的窗口），随意输入一些语句，语句间的单词以空格分隔。如下所示：
1. java python 2. ... 3. python scala hadoop
请确保输入的语句跨5秒钟。
3、切换到spark-shell所在终端窗口，查看打印出的结果，结果如下所示：
—————————————————————- Time: 1548752815000 ms —————————————————————- (java,1) (python,1) —————————————————————- Time: 1548752820000 ms —————————————————————- —————————————————————- Time: 1548752825000 ms —————————————————————- (python,2) (scala,1) (hadoop,1)
由上面的内容可以看出，第一次和第二次的数据进行了累加统计。
4、停止Spark流计算。在spark-shell窗口中，执行以下命令，停止流计算的执行：
1. ssc.stop(false)
或者，同时按下【Ctrl + C】键来终止。
