ToB企服应用市场:ToB评测及商务社交产业平台

标题: Spark Steaming有状态转换实行 [打印本页]

作者: 商道如狼道 时间: 2024-8-19 07:27
标题: Spark Steaming有状态转换实行
创建一个streaming目录

mkdir streaming

复制代码

一、运行网络版的WordCount

1. 连接虚拟机后利用sudo打开hosts后加入赤色方框内语句并生存：

sudo vim /etc/hosts

Netcat是一个用于TCP/UDP连接和监听的Linux工具, 重要用于网络传输及调试领域。先下载：

sudo apt-get update

复制代码

sudo apt-get -y install netcat-traditional

复制代码

2. 启动 NetCat 服务端，并在1234端口监听

nc -lk 1234 #在lsn下面输入nc –l –p 1234

利用xshell 打开一个新的选项卡，连接虚拟机。启动NetCat客户端，并连接Netcat服务端

nc localhost 1234

注意：假如客户端和服务端不在同一台机器，localhost 可以换成现实IP。

在服务端输入以下字符串，并按回车，可以在客户端收到消息，并打印出来。这里注意更换学号为你个人学号。

hello 你的学号

在客户端输入字符串，并按回车，可以在服务端收到消息，并打印出来。这里注意更换学号为你个人学号。

你好你的学号

在 NetCat 客户端的选项卡利用quit停止客户端历程。
利用有状态操作updateStateByKey实现词频统计。

在streaming目录下新建一个stateful目录，用于生存持久化的数据；接着编写独立的NetWordCountStateful.py代码

1	from pyspark import SparkContext
2	from pyspark.streaming import StreamingContext
3	sc = SparkContext("local[2]","NetworkWordCountStateful")
4	ssc = StreamingContext(sc,10)
5
6	ssc.checkpoint("file:///home/ubuntu/streaming/stateful")
7	def updateFunction(newValues, runningCount):
8	if runningCount is None:
9	runningCount = 0
10	return sum(newValues, runningCount)
11	lines = ssc.socketTextStream('localhost', 1234)
12	running_counts = lines.flatMap(lambda line:line.split(' ')).map(lambda xx,1)).updateStateByKey(updateFunction)
13	running_counts.pprint()
14	ssc.start()
15	ssc.awaitTermination()

注意：有状态转换必要举行设置检查点。

新建一个终端，开启服务端

nc -l -p 1234

再建一个“流计算”终端，运行NetWordCountStateful.py代码：

	cd $SPARK_HOME/bin
	spark-submit /路径/NetWordCountStateful.py

在 NetCat 服务端输入以下字符串，并按回车，观察Streaming WordCount的输出，并截图。

You jump I jump 1234

在客户端查察统计效果：

再次在服务端口输入以下字符串：

You and I jump 1234

回车后再次观察Streaming WordCount的输出，是否是累加后的效果。

可以用quit或ctrl+c停掉客户端，利用xshell的回滚来查察效果，由于回滚较快，以是在运行状态下查察效果截图较困难。

二、利用滑动窗口实现WordCount

1. 创建文件流监听目录：

mkdir logfile
cd logfile

在streaming目录下新建一个code文件夹，用于持久化数据的存储；然后新开一个终端，输入“pyspark”进入PySpark交互式情况后，输入以下代码：

from pyspark.streaming import StreamingContext
ssc = StreamingContext(sc,10)
ssc.checkpoint("file:///home/ubuntu/streaming/code")
lines = ssc.textFileStream("file:///home/ubuntu/streaming/logfile")
counts = lines.flatMap(lambda x:x.split(' ')).map(lambda x

x,1)).reduceByKeyAndWindow(lambda x, y:x+y, lambda x,y:x-y, 30, 10)
counts.pprint()
ssc.start()
ssc.awaitTermination()
输入ssc.start()后，程序就开始自动进入循环监听状态，如下图所示：

打开一个新的shell窗口，切换到logfile目录下，创建一个log.txt文档生存，再创建一个log_new.txt文档生存，里面输入一些随意的单词，并用空格间隔开。查察监听页面，可以看到打印效果，例如下图所示：

注：log.txt输入为“a b c a b c d”，log_new.txt输入为“a b d e f a b e f”。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)