Spark编程实验四：Spark Streaming编程

目录

一、目的与要求

二、实验内容

三、实验步骤

1、利用Spark Streaming对三种类型的基本数据源的数据进行处理

2、利用Spark Streaming对Kafka高级数据源的数据进行处理

3、完成DStream的两种有状态转换操作

4、把DStream的数据输出保存到文本文件或MySQL数据库中

四、结果分析与实验体会

一、目的与要求

1、通过实验掌握Spark Streaming的基本编程方法；
2、熟悉利用Spark Streaming处理来自不同数据源的数据。
3、熟悉DStream的各种转换操作。
4、熟悉把DStream的数据输出保存到文本文件或MySQL数据库中。

二、实验内容

1、参照教材示例，利用Spark Streaming对三种类型的基本数据源的数据进行处理。 
2、参照教材示例，完成kafka集群的配置，利用Spark Streaming对Kafka高级数据源的数据进行处理，注意topic为你的姓名全拼。
3、参照教材示例，完成DStream的两种有状态转换操作。
4、参照教材示例，完成把DStream的数据输出保存到文本文件或MySQL数据库中。

三、实验步骤

1、利用Spark Streaming对三种类型的基本数据源的数据进行处理

（1）文件流

首先打开第一个终端作为数据流终端，创建一个logfile目录：

[root@bigdata zhc]# cd /home/zhc/mycode/sparkstreaming
[root@bigdata sparkstreaming]# mkdir logfile
[root@bigdata sparkstreaming]# cd logfile

然后打开第二个终端作为流计算终端，在“/logfile/”目录下面新建一个py程序：

[root@bigdata logfile]# vim FileStreaming.py

输入如下代码：

#/home/zhc/mycode/sparkstreaming/logfile/FileStreaming.pyfrom pyspark import SparkContext, SparkConf
from pyspark.streaming import StreamingContextconf = SparkConf()
conf.setAppName('TestDStream')
conf.setMaster('local[2]')
sc = SparkContext(conf = conf)
ssc = StreamingContext(sc, 10)
lines = ssc.textFileStream('file:///home/zhc/mycode/sparkstreaming/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()

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

[root@bigdata logfile]# spark-submit FileStreaming.py

然后我们进入数据流终端，在logfile目录下新建一个log2.txt文件，然后往里面输入一些英文语句后保存退出，再次切换到流计算终端，就可以看见打印出单词统计信息了。

（2）套接字流

1）使用套接字流作为数据源

继续在流计算端的sparkstreaming目录下创建一个socket目录，然后在该目录下创建一个NetworkWordCount.py程序：

[root@bigdata sparkstreaming]# mkdir socket
[root@bigdata sparkstreaming]# cd socket
[root@bigdata socket]# vim NetworkWordCount.py

输入如下代码：

#/home/zhc/mycode/sparkstreaming/socket/NetworkWordCount.pyfrom __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, 5)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()

再在数据流终端启动Socket服务器端：

[root@bigdata logfile]# nc -lk 9999

然后再进入流计算终端，执行如下代码启动流计算：

[root@bigdata socket]# spark-submit NetworkWordCount.py localhost 9999

然后在数据流终端内手动输入一行英文句子后回车，多输入几次，流计算终端就会不断执行词频统计并打印出信息。

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

下面我们再前进一步，把数据源头的产生方式修改一下，不要使用nc程序，而是采用自己编写的程序产生Socket数据源。在数据流终端执行以下命令，编写DataSourceSocket.py文件：

[root@bigdata logfile]# cd /home/zhc/mycode/sparkstreaming/socket
[root@bigdata socket]# vim DataSourceSocket.py

输入如下代码：

#/home/zhc/mycode/sparkstreaming/socket/DataSourceSocket.py
import socket
# 生成socket对象
server = socket.socket()
# 绑定ip和端口
server.bind(('localhost', 9999))
# 监听绑定的端口
server.listen(1)
while 1:# 为了方便识别，打印一个“我在等待”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.')

继续在数据流终端执行如下命令启动Socket服务端：

[root@bigdata socket]# spark-submit DataSourceSocket.py

再进入流计算终端，执行如下代码启动流计算：

[root@bigdata socket]# spark-submit NetworkWordCount.py localhost 9999

（3）RDD队列流

继续在sparkstreaming目录下新建rddqueue目录并在该目录下创建RDDQueueStream.py程序：

[root@bigdata sparkstreaming]# mkdir rddqueue
[root@bigdata sparkstreaming]# cd rddqueue
[root@bigdata rddqueue]# vim RDDQueueStream.py

输入如下代码：

#/home/zhc/mycode/sparkstreaming/rddqueue/RDDQueueStreaming.py
import time
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
if __name__ == "__main__":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)

保存退出后，进入流计算终端再执行如下命令：

[root@bigdata rddqueue]# spark-submit RDDQueueStream.py

2、利用Spark Streaming对Kafka高级数据源的数据进行处理

此过程可以参照这篇博客的第四、五部分内容:

【数据采集与预处理】数据接入工具Kafka-CSDN博客icon-default.png?t=N7T8https://blog.csdn.net/Morse_Chen/article/details/135273370?spm=1001.2014.3001.5501

3、完成DStream的两种有状态转换操作

说明：上面的词频统计程序NetworkWordCount.py采取了无状态转换操作。

（1）滑动窗口转换操作

在socket目录下创建WindowedNetworkWordCount.py程序并输入如下代码：

#/home/zhc/mycode/sparkstreaming/socket/WindowedNetworkWordCount.py
from __future__ import print_function
import sys
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
if __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:///home/zhc/mycode/sparkstreaming/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程序：

[root@bigdata sparkstreaming]# cd /home/zhc/mycode/sparkstreaming/socket
[root@bigdata socket]# nc -lk 9999

然后再在流计算终端运行WindowedNetworkWordCount.py代码：

[root@bigdata socket]# spark-submit WindowedNetworkWordCount.py localhost 9999

这时，可以查看流计算终端内显示的词频动态统计结果，可以看到，随着时间的流逝，词频统计结果会发生动态变化。 

（2）updateStateByKey操作

在“/home/zhc/mycode/sparkstreaming/”路径下新建目录“/stateful”，并在该目录下新建代码文件NetworkWordCountStateful.py。

[root@bigdata sparkstreaming]# mkdir stateful
[root@bigdata sparkstreaming]# cd stateful
[root@bigdata stateful]# vim NetworkWordCountStateful.py

输入如下代码： 

#/home/zhc/mycode/sparkstreaming/stateful/NetworkWordCountStateful.py
from __future__ import print_function
import sys
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
if __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:///home/zhc/mycode/sparkstreaming/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程序：

[root@bigdata stateful]# nc  -lk  9999

在“流计算终端”，执行如下命令提交运行程序：

[root@bigdata stateful]# spark-submit NetworkWordCountStateful.py localhost 9999

在数据源终端内手动输入一些单词并回车，再切换到流计算终端，可以看到已经输出了类似如下的词频统计信息： 

4、把DStream的数据输出保存到文本文件或MySQL数据库中

（1）把DStream输出到文本文件中

在stateful目录下新建NetworkWordCountStatefulText.py文件：

[root@bigdata stateful]# vim NetworkWordCountStatefulText.py

输入如下代码： 

#/home/zhc/mycode/sparkstreaming/stateful/NetworkWordCountStatefulText.py
from __future__ import print_function
import sys
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
if __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:///home/zhc/mycode/sparkstreaming/stateful/statefultext")# 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.saveAsTextFiles("file:///home/zhc/mycode/sparkstreaming/stateful/statefultext/output")running_counts.pprint()ssc.start()ssc.awaitTermination()

在“数据源终端”，执行如下命令启动nc程序：

[root@bigdata stateful]# nc  -lk  9999

在“流计算终端”，执行如下命令提交运行程序：

[root@bigdata stateful]# spark-submit NetworkWordCountStatefulText.py localhost 9999

在数据源终端内手动输入一些单词并回车，再切换到流计算终端，可以看到已经输出了类似如下的词频统计信息：  

在“/home/zhc/mycode/sparkstreaming/stateful/statefultext”目录下便可查看到如下输出目录结果：

进入某个目录下，就可以看到类似part-00000的文件，里面包含了流计算过程的输出结果。

（2）把DStream写入到MySQL数据库中

首先启动MySQL数据库：

[root@bigdata stateful]# systemctl start mysqld.service
[root@bigdata stateful]# mysql -u root -p

然后创建spark数据库和wordcount表：

mysql> use spark;
mysql> create table wordcount (word char(20), count int(4));

然后再在终端安装python连接MySQL的模块：

[root@bigdata stateful]# pip3 install PyMySQL

在stateful目录并在该目录下创建NetworkWordCountStatefulDB.py文件： 

[root@bigdata stateful]# vim NetworkWordCountStatefulDB.py

输入如下代码： 

#/home/zhc/mycode/sparkstreaming/stateful/NetworkWordCountStatefulDB.py
from __future__ import print_function 
import sys 
import pymysql 
from pyspark import SparkContext
from pyspark.streaming import StreamingContext 
if __name__ == "__main__":if len(sys.argv) != 3:print("Usage: NetworkWordCountStateful <hostname> <port>", file=sys.stderr)exit(-1)sc = SparkContext(appName="PythonStreamingStatefulNetworkWordCount")ssc = StreamingContext(sc, 1)ssc.checkpoint("file:///home/zhc/mycode/sparkstreaming/stateful/statefuldb")  # 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() def dbfunc(records):db = pymysql.connect(host="localhost",user="root",password="MYsql123!",database="spark")cursor = db.cursor() def doinsert(p):sql = "insert into wordcount(word,count) values ('%s', '%s')" % (str(p[0]), str(p[1]))try:cursor.execute(sql)db.commit()except:db.rollback()for item in records:doinsert(item) def func(rdd):repartitionedRDD = rdd.repartition(3)repartitionedRDD.foreachPartition(dbfunc)running_counts.foreachRDD(func)ssc.start()ssc.awaitTermination()

在“数据源终端”，执行如下命令启动nc程序：

[root@bigdata stateful]# nc  -lk  9999

在“流计算终端”，执行如下命令提交运行程序：

[root@bigdata stateful]# spark-submit NetworkWordCountStatefulDB.py localhost 9999

在数据源终端内手动输入一些单词并回车，再切换到流计算终端，可以看到已经输出了类似如下的词频统计信息：

 

到MySQL终端便可以查看wordcount表中的内容：

mysql> select * from wordcount;

.......

四、结果分析与实验体会

        Spark Streaming是一个用于实时数据处理的流式计算框架，它基于 Apache Spark 平台，提供了高可靠性、高吞吐量和容错性强等特点。在进行 Spark Streaming 编程的实验中，掌握了Spark Streaming的基本编程方法；能够利用Spark Streaming处理来自不同数据源的数据以及DStream的各种转换操作；把DStream的数据输出保存到文本文件或MySQL数据库中。
        理解DStream：DStream 是 Spark Streaming 的核心概念，代表连续的数据流。在编程时，我们可以通过输入源（比如 Kafka、Flume、HDFS）创建一个 DStream 对象，并对其进行转换和操作。需要注意的是，DStream 是以时间片为单位组织数据的，因此在编写代码时要考虑时间窗口的大小和滑动间隔。
        适当设置批处理时间间隔：批处理时间间隔决定了 Spark Streaming 处理数据的粒度，过小的时间间隔可能导致频繁的任务调度和资源开销，而过大的时间间隔则可能造成数据处理延迟。因此，在实验中需要根据具体场景和需求来选择合适的时间间隔。
        使用合适的转换操作：Spark Streaming 提供了丰富的转换操作，如 map、flatMap、filter、reduceByKey 等，可以实现对数据流的转换和处理。在实验中，需要根据具体业务逻辑和需求选择合适的转换操作，并合理组合这些操作，以获取期望的结果。
        考虑容错性和数据丢失：Spark Streaming 具备很好的容错性，可以通过记录数据流的偏移量来保证数据不会丢失。在实验中，需要注意配置合适的容错机制，确保数据处理过程中的异常情况能够被恢复，并尽量避免数据丢失。
        优化性能和资源利用：对于大规模的实时数据处理任务，性能和资源利用是非常重要的。在实验中，可以通过调整并行度、合理设置缓存策略、使用广播变量等手段来提高性能和资源利用效率。
        总的来说，Spark Streaming 是一个功能强大且易用的流式计算框架，通过合理使用其提供的特性和操作，可以实现各种实时数据处理需求。在实验中，需要深入理解其原理和机制，并根据具体需求进行合理配置和优化，以获得良好的性能和结果。

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

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

相关文章

NModbus-一个C#的Modbus协议库实现

NModbus-一个C#的Modbus协议库实现

NModbus-一个基于C#实现的Modbus通信协议库 最近在学习C#的时候&#xff0c;因为之前做过环保设备时使用C做过环保设备采集使用到了Modbus协议&#xff0c;当时看了一下基于C语言开发的libmodbus库。所以特意搜索看了一下C#下有什么Modbus协议库&#xff0c;在Github上面找了一…
阅读更多...
[蓝桥杯2020国赛]答疑

[蓝桥杯2020国赛]答疑

答疑 题目描述 有 n 位同学同时找老师答疑。每位同学都预先估计了自己答疑的时间。 老师可以安排答疑的顺序&#xff0c;同学们要依次进入老师办公室答疑。 一位同学答疑的过程如下&#xff1a; 首先进入办公室&#xff0c;编号为 i 的同学需要 si​ 毫秒的时间。然后同学问…
阅读更多...
After Effects 2021 for Mac(AE 2021)

After Effects 2021 for Mac(AE 2021)

After Effects 2021是一款由Adobe公司开发的视频特效和动态图形制作软件&#xff0c;它主要用于电影、电视和网络视频的后期制作。该软件可以帮助用户创建各种令人惊叹的视觉效果&#xff0c;包括动态图形、文字特效、粒子系统、3D渲染等。 After Effects 2021提供了数百种特效…
阅读更多...
dash 中的模式匹配回调函数Pattern-Matching Callbacks 8

dash 中的模式匹配回调函数Pattern-Matching Callbacks 8

模式匹配 模式匹配回调选择器 MATCH、ALL 和 ALLSMALLER 允许您编写可以响应或更新任意或动态数量组件的回调函数。 此示例呈现任意数量的 dcc. Dropdown 元素&#xff0c;并且只要任何 dcc. Dropdown 元素发生更改&#xff0c;就会触发回调。尝试添加几个下拉菜单并选择它们的…
阅读更多...
TCP/IP的五层网络模型

TCP/IP的五层网络模型

目录 封装&#xff08;打包快递&#xff09; 6.1应用层 6.2传输层 6.3网络层 6.4数据链路层 6.5物理层 分用&#xff08;拆快递&#xff09; 6.5物理层 6.4数据链路层 6.3网络层 6.2传输层 6.1应用层 封装&#xff08;打包快递&#xff09; 6.1应用层 此时做的数据…
阅读更多...
Android Studio配置国内镜像源和HTTP代理/解决：Android Studio下载gradle速度慢的问题

Android Studio配置国内镜像源和HTTP代理/解决：Android Studio下载gradle速度慢的问题

&#xff08;方案一&#xff09;Android Studio配置国内镜像源和HTTP代理 一、配置国内镜像源/依赖库 1.1 打开项目的setting.gradle.kts文件 配置进去 pluginManagement {repositories {maven { urluri ("https://www.jitpack.io")}maven { urluri ("https:…
阅读更多...
【C语言】分支与循环语句

【C语言】分支与循环语句

什么是语句&#xff1f; C语句可分为以下五类&#xff1a; 表达式语句函数调用语句控制语句 &#xff08;本篇重点介绍&#xff09;复合语句空语句 控制语句用于控制程序的执行流程&#xff0c;以实现程序的各种结构方式。C语言支持三种结构&#xff1a; 顺序结构选择结构循…
阅读更多...
第一讲：BeanFactory和ApplicationContext

第一讲：BeanFactory和ApplicationContext

BeanFactory和ApplicationContext 什么是BeanFactory 它是ApplicationContext的父接口它才是Spring的核心容器&#xff0c;主要的ApplicationContext实现都组合了它的功能 BeanFactory能做什么? 表面上看BeanFactory的主要方法只有getBean()&#xff0c;实际上控制反转、基…
阅读更多...
产品经理学习-策略产品指标

产品经理学习-策略产品指标

目录&#xff1a; 数据指标概述 通用指标介绍 Web端常用指标 移动端常用指标 如何选择一个合适的数据指标 数据指标概述 指标是衡量目标的一个参数&#xff0c;指一项活动中预期达到的指标、目标等&#xff0c;一般用数据表示&#xff0c;因此又称为数据指标&#xff1b;…
阅读更多...
pytorch03：transforms常见数据增强操作

pytorch03：transforms常见数据增强操作

目录 一、数据增强二、transforms--Crop裁剪2.1 transforms.CenterCrop2.2 transforms.RandomCrop2.3 RandomResizedCrop2.4 FiveCrop和TenCrop 三、transforms—Flip翻转、旋转3.1RandomHorizontalFlip和RandomVerticalFlip3.2 RandomRotation 四、transforms —图像变换4.1 t…
阅读更多...
Github项目推荐-vocal-separate

Github项目推荐-vocal-separate

项目地址 vocal-separate: 项目简述 这是一个音乐和人声分离的项目&#xff0c;基于python开发。有图形化操作界面&#xff0c;看起来还不错。 项目截图
阅读更多...
二叉树BFS

二叉树BFS

前置知识 二叉树节点的定义 二叉树是递归定义的 /*** Definition for a binary tree node.&#xff08;LeetCode&#xff09;*/public class TreeNode {int val;TreeNode left;TreeNode right;TreeNode() {}TreeNode(int val) { this.val val; }TreeNode(int val, TreeNode…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023