大数据分析与应用实验任务十一

大数据分析与应用实验任务十一

实验目的

  • 通过实验掌握spark Streaming相关对象的创建方法；

  • 熟悉spark Streaming对文件流、套接字流和RDD队列流的数据接收处理方法；

  • 熟悉spark Streaming的转换操作，包括无状态和有状态转换。

  • 熟悉spark Streaming输出编程操作。

实验任务

一、DStream 操作概述

  1. 创建 StreamingContext 对象

    登录 Linux 系统后，启动 pyspark。进入 pyspark 以后，就已经获得了一个默认的 SparkConext 对象，也就是 sc。因此，可以采用如下方式来创建 StreamingContext 对象：

    from pyspark.streaming import StreamingContext 
sscluozhongye = StreamingContext(sc, 1)

    image-20231207112253827

    如果是编写一个独立的 Spark Streaming 程序，而不是在 pyspark 中运行，则需要在代码文件中通过类似如下的方式创建 StreamingContext 对象：

    from pyspark import SparkContext, SparkConf 
from pyspark.streaming import StreamingContext 
conf = SparkConf() 
conf.setAppName('TestDStream') 
conf.setMaster('local[2]') 
sc = SparkContext(conf = conf) 
ssc = StreamingContext(sc, 1)
print("创建成功,lzy防伪")

    image-20231207112652285

二、基本输入源
  1. 文件流

  • 在 pyspark 中创建文件流

    首先，在 Linux 系统中打开第 1 个终端（为了便于区分多个终端，这里记作“数据源终端”），创建一个 logfile 目录，命令如下：

    cd /root/Desktop/luozhongye/
mkdir streaming 
cd streaming 
mkdir logfile

    image-20231207112923323

    其次，在 Linux 系统中打开第二个终端（记作“流计算终端”），启动进入 pyspark，然后，依次输入如下语句：

    from pyspark import SparkContext 
from pyspark.streaming import StreamingContext 
ssc = StreamingContext(sc, 10) 
lines = ssc.textFileStream('file:///root/Desktop/luozhongye/streaming/logfile') 
words = lines.flatMap(lambda line: line.split(' ')) 
wordCounts = words.map(lambda x : (x,1)).reduceByKey(lambda a,b:a+b) 
wordCounts.pprint() 
ssc.start() 
ssc.awaitTermination()

image-20231207113305405

  • 采用独立应用程序方式创建文件流

    #!/usr/bin/env python3 
from pyspark import SparkContext, SparkConf 
from pyspark.streaming import StreamingContext 
conf = SparkConf() 
conf.setAppName('TestDStream') 
conf.setMaster('local[2]') 
sc = SparkContext(conf = conf) 
ssc = StreamingContext(sc, 10) 
lines = ssc.textFileStream('file:///root/Desktop/luozhongye/streaming/logfile') 
words = lines.flatMap(lambda line: line.split(' ')) 
wordCounts = words.map(lambda x : (x,1)).reduceByKey(lambda a,b:a+b) 
wordCounts.pprint() 
ssc.start() 
ssc.awaitTermination()
print("2023年12月7日lzy")

    保存该文件，并执行以下命令：

    cd /root/Desktop/luozhongye/streaming/logfile/ 
spark-submit FileStreaming.py

image-20231207114014647

  1. 套接字流

  • 使用套接字流作为数据源

    新建一个代码文件“/root/Desktop/luozhongye/streaming/socket/NetworkWordCount.py”，在NetworkWordCount.py 中输入如下内容：

    #!/usr/bin/env python3 
from __future__ import print_function
import sys
from pyspark import SparkContext
from pyspark.streaming import StreamingContextif __name__ == "__main__":if len(sys.argv) != 3:print("Usage: NetworkWordCount.py <hostname> <port>", file=sys.stderr)exit(-1)sc = SparkContext(appName="PythonStreamingNetworkWordCount")ssc = StreamingContext(sc, 1)lines = ssc.socketTextStream(sys.argv[1], int(sys.argv[2]))counts = lines.flatMap(lambda line: line.split(" ")).map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b)counts.pprint()ssc.start()ssc.awaitTermination()

    使用如下 nc 命令生成一个 Socket 服务器端：

    nc -lk 9999

    新建一个终端（记作“流计算终端”），执行如下代码启动流计算：

    cd /root/Desktop/luozhongye/streaming/socket 
/usr/local/spark/bin/spark-submit NetworkWordCount.py localhost 9999

image-20231208002212790

  • 使用 Socket 编程实现自定义数据源

    新建一个代码文件“/root/Desktop/luozhongye/streaming/socket/DataSourceSocket.py”，在 DataSourceSocket.py 中输入如下代码：

    #!/usr/bin/env python3 
import socket# 生成 socket 对象
server = socket.socket()
# 绑定 ip 和端口
server.bind(('localhost', 9999))
# 监听绑定的端口
server.listen(1)
while 1:# 为了方便识别，打印一个“I’m waiting the connect...”print("I'm waiting the connect...")# 这里用两个值接收，因为连接上之后使用的是客户端发来请求的这个实例# 所以下面的传输要使用 conn 实例操作conn, addr = server.accept()# 打印连接成功print("Connect success! Connection is from %s " % addr[0])# 打印正在发送数据print('Sending data...')conn.send('I love hadoop I love spark hadoop is good spark is fast'.encode())conn.close()print('Connection is broken.')
print("2023年12月7日lzy")

    执行如下命令启动 Socket 服务器端：

    cd /root/Desktop/luozhongye/streaming/socket 
/usr/local/spark/bin/spark-submit DataSourceSocket.py

    新建一个终端（记作“流计算终端”），输入以下命令启动 NetworkWordCount 程序：

    cd /root/Desktop/luozhongye/streaming/socket 
/usr/local/spark/bin/spark-submit NetworkWordCount.py localhost 9999

image-20231208003303167

  1. RDD 队列流

    Linux 系统中打开一个终端，新建一个代码文件“/root/Desktop/luozhongye/ streaming/rddqueue/ RDDQueueStream.py”，输入以下代码：

    #!/usr/bin/env python3 
import time
from pyspark import SparkContext
from pyspark.streaming import StreamingContextif __name__ == "__main__":print("")sc = SparkContext(appName="PythonStreamingQueueStream")ssc = StreamingContext(sc, 2)# 创建一个队列，通过该队列可以把 RDD 推给一个 RDD 队列流rddQueue = []for i in range(5):rddQueue += [ssc.sparkContext.parallelize([j for j in range(1, 1001)], 10)]time.sleep(1)# 创建一个 RDD 队列流inputStream = ssc.queueStream(rddQueue)mappedStream = inputStream.map(lambda x: (x % 10, 1))reducedStream = mappedStream.reduceByKey(lambda a, b: a + b)reducedStream.pprint()ssc.start()ssc.stop(stopSparkContext=True, stopGraceFully=True)

    下面执行如下命令运行该程序：

    cd /root/Desktop/luozhongye/streaming/rddqueue 
/usr/local/spark/bin/spark-submit RDDQueueStream.py

image-20231208004439462

三、转换操作

  1. 滑动窗口转换操作

    对“套接字流”中的代码 NetworkWordCount.py 进行一个小的修改，得到新的代码文件“/root/Desktop/luozhongye/streaming/socket/WindowedNetworkWordCount.py”，其内容如下：

    #!/usr/bin/env python3 
from __future__ import print_function
import sys
from pyspark import SparkContext
from pyspark.streaming import StreamingContextif __name__ == "__main__":if len(sys.argv) != 3:print("Usage: WindowedNetworkWordCount.py <hostname> <port>", file=sys.stderr)exit(-1)sc = SparkContext(appName="PythonStreamingWindowedNetworkWordCount")ssc = StreamingContext(sc, 10)ssc.checkpoint("file:///root/Desktop/luozhongye/streaming/socket/checkpoint")lines = ssc.socketTextStream(sys.argv[1], int(sys.argv[2]))counts = lines.flatMap(lambda line: line.split(" ")) \.map(lambda word: (word, 1)) \.reduceByKeyAndWindow(lambda x, y: x + y, lambda x, y: x - y, 30, 10)counts.pprint()ssc.start()ssc.awaitTermination()

为了测试程序的运行效果，首先新建一个终端（记作“数据源终端”），执行如下命令运行nc 程序：

   cd /root/Desktop/luozhongye/streaming/socket/ nc -lk 9999

然后，再新建一个终端（记作“流计算终端”），运行客户端程序 WindowedNetworkWordCount.py，命令如下：

   cd /root/Desktop/luozhongye/streaming/socket/ /usr/local/spark/bin/spark-submit WindowedNetworkWordCount.py localhost 9999

在数据源终端内，连续输入 10 个“hadoop”，每个 hadoop 单独占一行（即每输入一个 hadoop就按回车键），再连续输入 10 个“spark”，每个 spark 单独占一行。这时，可以查看流计算终端内显示的词频动态统计结果，可以看到，随着时间的流逝，词频统计结果会发生动态变化。

image-20231208005821701

  1. updateStateByKey 操作

    在“/root/Desktop/luozhongye/streaming/stateful/”目录下新建一个代码文件 NetworkWordCountStateful.py，输入以下代码：

    #!/usr/bin/env python3 
from __future__ import print_function
import sys
from pyspark import SparkContext
from pyspark.streaming import StreamingContextif __name__ == "__main__":if len(sys.argv) != 3:print("Usage: NetworkWordCountStateful.py <hostname> <port>", file=sys.stderr)exit(-1)sc = SparkContext(appName="PythonStreamingStatefulNetworkWordCount")ssc = StreamingContext(sc, 1)ssc.checkpoint("file:///root/Desktop/luozhongye/streaming/stateful/")# RDD with initial state (key, value) pairsinitialStateRDD = sc.parallelize([(u'hello', 1), (u'world', 1)])def updateFunc(new_values, last_sum):return sum(new_values) + (last_sum or 0)lines = ssc.socketTextStream(sys.argv[1], int(sys.argv[2]))running_counts = lines.flatMap(lambda line: line.split(" ")) \.map(lambda word: (word, 1)) \.updateStateByKey(updateFunc, initialRDD=initialStateRDD)running_counts.pprint()ssc.start()ssc.awaitTermination()

    新建一个终端（记作“数据源终端”），执行如下命令启动 nc 程序：

    nc -lk 9999

    新建一个 Linux 终端（记作“流计算终端”），执行如下命令提交运行程序：

    cd /root/Desktop/luozhongye/streaming/stateful 
/usr/local/spark/bin/spark-submit NetworkWordCountStateful.py localhost 9999

image-20231208010814959

四、把 DStream 输出到文本文件中

下面对之前已经得到的“/root/Desktop/luozhongye/streaming/stateful/NetworkWordCountStateful.py”代码进行简单的修改，把生成的词频统计结果写入文本文件中。

修改后得到的新代码文件“/root/Desktop/luozhongye/streaming/stateful/NetworkWordCountStatefulText.py”的内容如下：

#!/usr/bin/env python3 
from __future__ import print_function
import sys
from pyspark import SparkContext
from pyspark.streaming import StreamingContextif __name__ == "__main__":if len(sys.argv) != 3:print("Usage: NetworkWordCountStateful.py <hostname> <port>", file=sys.stderr)exit(-1)sc = SparkContext(appName="PythonStreamingStatefulNetworkWordCount")ssc = StreamingContext(sc, 1)ssc.checkpoint("file:///root/Desktop/luozhongye/streaming/stateful/")# RDD with initial state (key, value) pairs initialStateRDD = sc.parallelize([(u'hello', 1), (u'world', 1)])def updateFunc(new_values, last_sum):return sum(new_values) + (last_sum or 0)lines = ssc.socketTextStream(sys.argv[1], int(sys.argv[2]))running_counts = lines.flatMap(lambda line: line.split(" ")) \.map(lambda word: (word, 1)) \.updateStateByKey(updateFunc, initialRDD=initialStateRDD)running_counts.saveAsTextFiles("file:///root/Desktop/luozhongye/streaming/stateful/output")running_counts.pprint()ssc.start()ssc.awaitTermination()

新建一个终端（记作“数据源终端”），执行如下命令运行nc 程序：

cd /root/Desktop/luozhongye/streaming/socket/ 
nc -lk 9999

新建一个 Linux 终端（记作“流计算终端”），执行如下命令提交运行程序：

cd /root/Desktop/luozhongye/streaming/stateful 
/usr/local/spark/bin/spark-submit NetworkWordCountStatefulText.py localhost 9999

image-20231208012123002

实验心得

通过本次实验，我深入理解了Spark Streaming，包括创建StreamingContext、DStream等对象。同时，我了解了Spark Streaming对不同类型数据流的处理方式，如文件流、套接字流和RDD队列流。此外，我还熟悉了Spark Streaming的转换操作和输出编程操作，并掌握了map、flatMap、filter等方法。最后，我能够自定义输出方式和格式。总之，这次实验让我全面了解了Spark Streaming，对未来的学习和工作有很大的帮助。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.rhkb.cn/news/215004.html

如若内容造成侵权/违法违规/事实不符，请联系长河编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

极简模式，助力宏观数据监控

极简模式，助力宏观数据监控

随着UWA GOT Online采样的参数越来越多样化&#xff0c;为了提升开发者的使用体验&#xff0c;我们最新推出了三种预设数据采集方案&#xff1a;极简模式、CPU模式、内存模式。该更新旨在降低多数据采集对数据准确性的干扰&#xff0c;同时也为大家提供更精准且有针对性的数据指…
阅读更多...
mac苹果笔记本电脑如何强力删除卸载app软件？

mac苹果笔记本电脑如何强力删除卸载app软件？

苹果电脑怎样删除app&#xff1f;不是把app移到废纸篓就行了吗&#xff0c;十分简单呢&#xff01; 其实不然&#xff0c;因为在Mac电脑上&#xff0c;删除应用程序只是删除了应用程序的主要组件。大多数时候&#xff0c;系统会有一个相当长的目录&#xff0c;包含所有与应用程…
阅读更多...
vuepress-----22、其他评论方案

vuepress-----22、其他评论方案

vuepress 支持评论 本文讲述 vuepress 站点如何集成评论系统&#xff0c;选型是 valineleancloud, 支持匿名评论&#xff0c;缺点是数据没有存储在自己手里。市面上也有其他的方案, 如 gitalk,vssue 等, 但需要用户登录 github 才能发表评论, 但 github 经常无法连接,导致体验…
阅读更多...
搞定这些软件测试面试题，面试通过率提高百分之80！

搞定这些软件测试面试题，面试通过率提高百分之80！

十九、持续集成 19.1 jenkins ant jmeter svn 接口自动化测试? jenkins ant jmeter svn 环境搭建 原来这个环境是我这边搭建的&#xff0c; 主要是几个步骤&#xff0c; 第一 Jenkins 安装、第二&#xff0c;ant 安装、第三&#xff0c; jmeter 安装、第四&#xff0c; …
阅读更多...
关于加密解密，加签验签那些事

关于加密解密，加签验签那些事

面对MD5、SHA、DES、AES、RSA等等这些名词你是否有很多问号&#xff1f;这些名词都是什么&#xff1f;还有什么公钥加密、私钥解密、私钥加签、公钥验签。这些都什么鬼&#xff1f;或许在你日常工作没有听说过这些名词&#xff0c;但是一旦你要设计一个对外访问的接口&#xff…
阅读更多...
API测试基础之http协议

API测试基础之http协议

http简介&#xff1a; http&#xff08;超文本传输协议&#xff09;是一个简单的请求-响应协议&#xff0c;它通常运行在TCP&#xff08;传输控制协议&#xff09;之上。它指定了客户端可能发送给服务器什么样的消息以及得到什么样的响应。请求和响应消息的头以ASCII码形式给出…
阅读更多...
生成式人工智能笔记-AIGC笔记

生成式人工智能笔记-AIGC笔记

生成式人工智能笔记-AIGC笔记 十多年前&#xff0c;人工智能还只是一个不被人看好的小众领域&#xff0c;但是现在&#xff0c;它却已经成了街头巷尾的热点谈资&#xff0c;几乎任何事情都可以和人工智能联系在一起。 人工智能包括基础层、技术层和应用层。 基础层是人工智能…
阅读更多...
WebRTC AEC回声消除算法拆解

WebRTC AEC回声消除算法拆解

WebRTC AEC算法流程分析——时延估计&#xff08;一&#xff09; 其实&#xff0c;网上有很多类似资料&#xff0c;各个大厂研发不同应用场景设备的音频工程师基本都对其进行了拆解&#xff0c;有些闪烁其词&#xff0c;有些却很深奥&#xff0c;笔者随着对WebRTC了解的深入&a…
阅读更多...
scripty妙用

scripty妙用

在monorepo项目中&#xff0c;随着子模块增多&#xff0c; 每个子项目都需要配置各自的package.json,并且大同小异&#xff0c;为了进一步提高配置效率&#xff0c;引入了scripty&#xff0c;自己写脚本&#xff0c;直接就可以用哦 1、安装 npm install scripty --save-dev 2…
阅读更多...
实现安装“自由化”！在Windows 11中如何绕过“您尝试安装的应用程序未通过微软验证”

实现安装“自由化”！在Windows 11中如何绕过“您尝试安装的应用程序未通过微软验证”

这篇文章描述了如果你不能安装应用程序，而是当你在Windows 11中看到消息“您尝试安装的应用程序未通过微软验证”时该怎么办。完成这些步骤将取消你安装的应用程序必须经过Microsoft验证的要求。 使用设置应用程序 “设置”应用程序提供了绕过此警告消息的最简单方法，以便你…
阅读更多...
基于JavaWeb+SSM+Vue马拉松报名系统微信小程序的设计和实现

基于JavaWeb+SSM+Vue马拉松报名系统微信小程序的设计和实现

基于JavaWebSSMVue马拉松报名系统微信小程序的设计和实现 源码获取入口Lun文目录前言主要技术系统设计功能截图订阅经典源码专栏Java项目精品实战案例《500套》 源码获取 源码获取入口 Lun文目录 1系统概述 1 1.1 研究背景 1 1.2研究目的 1 1.3系统设计思想 1 2相关技术 2 2.…
阅读更多...
记录 | vscode设置自动换行

记录 | vscode设置自动换行

右上菜单栏 -> 查看 -> 打开自动换行 或者还有种方式&#xff0c;如下&#xff0c; 左下角小齿轮&#xff0c;点击设置 然后输入 Editor: Word Wrap &#xff0c;把开关打开为 on
阅读更多...
微信小程序 长按录音+录制视频

微信小程序 长按录音+录制视频

<view class"bigCircle" bindtouchstart"start" bindtouchend"stop"><view class"smallCircle {{startVedio?onVedio:}}"><text>{{startVedio?正在录音:长按录音}}</text></view> </view> <…
阅读更多...
AWTK 串口屏开发(1) - Hello World

AWTK 串口屏开发(1) - Hello World

1. 功能 这个例子很简单&#xff0c;制作一个调节温度的界面。在这里例子中&#xff0c;模型&#xff08;也就是数据&#xff09;里只有一个温度变量&#xff1a; 变量名数据类型功能说明温度整数温度。范围 (0-100) 摄氏度 2. 创建项目 从模板创建项目&#xff0c;将 hmi/…
阅读更多...
adb命令学习记录

adb命令学习记录

1、 adb ( android debug bridge)安卓调试桥&#xff0c;用于完成电脑和手机之间的通信控制。 xcode来完成对于ios设备的操控&#xff0c;前提是有个mac电脑。 安卓系统是基于linux内核来进行开发的。 2、adb的安装: 本身 adb是 android SDK 其中自带的工具&#xff0c;用于完…
阅读更多...
【无标题】从0到1 搭建一个vue3+Django项目

【无标题】从0到1 搭建一个vue3+Django项目

目录 一、后端项目python django二、前端项目vitevue3三、后端配置3.1 将路由指向app3.2 app下创建urls.py&#xff0c; 写入路由3.3 views写入test函数3.4 启动服务&#xff0c;访问路由 四、前端配置4.1 安装一些工具库及创建文件4.1.1 安装需要用的三方库4.1.2 创建文件 4.2…
阅读更多...
尝试通过AI模型进行简单的编码

尝试通过AI模型进行简单的编码

一、前言 最近尝试通过AI来编程&#xff0c;总体感觉还是能处理写简单的问题&#xff0c;复杂的问题目前还是无法解决。主要的痛点还是数据噪音&#xff0c;就是AI永远不会承认它不会&#xff0c;它会给你的一个错误的信息&#xff0c;它也不会告诉你你的问题它暂时无法完整正…
阅读更多...
JUC并发编程03——LockSupport与线程中断

JUC并发编程03——LockSupport与线程中断

一.线程中断机制 假设从网络下载一个100M的文件&#xff0c;如果网速很慢&#xff0c;用户等得不耐烦&#xff0c;就可能在下载过程中点“取消”&#xff0c;这时&#xff0c;程序就需要中断下载线程的执行。 1.1如何停止中断运行中的线程&#xff1f; 通过一个volatile变量…
阅读更多...
智能优化算法应用：基于郊狼算法3D无线传感器网络(WSN)覆盖优化 - 附代码

智能优化算法应用：基于郊狼算法3D无线传感器网络(WSN)覆盖优化 - 附代码

智能优化算法应用&#xff1a;基于郊狼算法3D无线传感器网络(WSN)覆盖优化 - 附代码 文章目录 智能优化算法应用&#xff1a;基于郊狼算法3D无线传感器网络(WSN)覆盖优化 - 附代码1.无线传感网络节点模型2.覆盖数学模型及分析3.郊狼算法4.实验参数设定5.算法结果6.参考文献7.MA…
阅读更多...
ES6原生音乐播放器（有接口）

ES6原生音乐播放器（有接口）

视频展示 ES6音乐播放器 项目介绍 GutHub地址&#xff1a;GitHub - baozixiangqianchong/ES6_MusicPlayer: 音乐播放器 ES6_MusicPlayer 是基于JavaScriptES6Ajax等通过原生构建的项目。能够充分锻炼JS能力。 本项目有主页、详情页、歌单页面三部分组成 ├── assets&…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023