创建一个streaming目录
一、运行网络版的WordCount
1. 连接虚拟机后利用sudo打开hosts后加入赤色方框内语句并生存：
sudo vim /etc/hosts
Netcat是一个用于TCP/UDP连接和监听的Linux工具, 重要用于网络传输及调试领域。先下载：
- 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 x  x,1)).updateStateByKey(updateFunction)
| | 13
| running_counts.pprint()
| | 14
| ssc.start()
| | 15
| ssc.awaitTermination()
|
注意：有状态转换必要举行设置检查点。
新建一个终端，开启服务端
再建一个“流计算”终端，运行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 xx,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企服之家，中国第一个企服评测及商务社交产业平台。 | 
