Spark编程实验二:RDD编程初级实践

目录

一、目的与要求

二、实验内容

三、实验步骤

1、pyspark交互式编程

2、编写独立应用程序实现数据去重

3、编写独立应用程序实现求平均值问题

4、三个综合实例

四、结果分析与实验体会


一、目的与要求

1、熟悉Spark的RDD基本操作及键值对操作;
2、熟悉使用RDD编程解决实际具体问题的方法。

二、实验内容

1、pyspark交互式编程

给定数据集 data1.txt,包含了某大学计算机系的成绩,数据格式如下所示:

Tom,DataBase,80

Tom,Algorithm,50

Tom,DataStructure,60

Jim,DataBase,90

Jim,Algorithm,60

Jim,DataStructure,80

……

请根据给定的实验数据,在pyspark中通过编程来计算以下内容:

(1)该系总共有多少学生;
(2)该系共开设了多少门课程;
(3)Tom同学的总成绩平均分是多少;
(4)求每名同学的选修的课程门数;
(5)该系DataBase课程共有多少人选修;
(6)各门课程的平均分是多少;
(7)使用累加器计算共有多少人选了DataBase这门课。

2、编写独立应用程序实现数据去重

        对于两个输入文件A和B,编写Spark独立应用程序,对两个文件进行合并,并剔除其中重复的内容,得到一个新文件C。下面是输入文件和输出文件的一个样例,供参考。

        输入文件A的样例如下:
        20170101    x
        20170102    y
        20170103    x
        20170104    y
        20170105    z
        20170106    z

        输入文件B的样例如下:
        20170101    y
        20170102    y
        20170103    x
        20170104    z
        20170105    y

        根据输入的文件A和B合并得到的输出文件C的样例如下:
        20170101    x
        20170101    y
        20170102    y
        20170103    x
        20170104    y
        20170104    z
        20170105    y
        20170105    z
        20170106    z

3、编写独立应用程序实现求平均值问题

        每个输入文件表示班级学生某个学科的成绩,每行内容由两个字段组成,第一个是学生名字,第二个是学生的成绩;编写Spark独立应用程序求出所有学生的平均成绩,并输出到一个新文件中。下面是输入文件和输出文件的一个样例,供参考。

        Algorithm成绩:
        小明 92
        小红 87
        小新 82
        小丽 90

        Database成绩:
        小明 95
        小红 81
        小新 89
        小丽 85

        Python成绩:
        小明 82
        小红 83
        小新 94
        小丽 91

        平均成绩如下:
        (小红,83.67)
        (小新,88.33)
        (小明,89.67)
        (小丽,88.67)

4、三个综合实例

        题目详情可查看实验步骤。

三、实验步骤

1、pyspark交互式编程

先在终端启动pyspark:

[root@bigdata zhc]# pyspark

(1)该系总共有多少学生;

>>> lines = sc.textFile("file:///home/zhc/datasets/data1.txt")
>>> res = lines.map(lambda x:x.split(",")).map(lambda x: x[0])     # 获取每行数据的第1列
>>> distinct_res = res.distinct()         # 去重操作
>>> distinct_res.count()        # 取元素总个数

执行结果: 

(2)该系共开设了多少门课程;

>>> lines = sc.textFile("file:///home/zhc/datasets/data1.txt")
>>> res = lines.map(lambda x:x.split(",")).map(lambda x:x[1])      # 获取每行数据的第2列
>>> distinct_res = res.distinct()         # 去重操作
>>> distinct_res.count()        # 取元素总个数

执行结果:

(3)Tom同学的总成绩平均分是多少;

>>> lines = sc.textFile("file:///home/zhc/datasets/data1.txt")
>>> res = lines.map(lambda x:x.split(",")).filter(lambda x:x[0]=="Tom")    # 筛选Tom同学的成绩信息
>>> res.foreach(print) 
>>> score = res.map(lambda x:int(x[2]))       # 提取Tom同学的每门成绩,并转换为int类型
>>> num = res.count()        # Tom同学选课门数
>>> sum_score = score.reduce(lambda x,y:x+y)       # Tom同学的总成绩
>>> avg = sum_score/num       # 总成绩/门数=平均分
>>> print(avg)

执行结果:

(4)求每名同学的选修的课程门数;

>>> lines = sc.textFile("file:///home/zhc/datasets/data1.txt")
>>> res = lines.map(lambda x:x.split(",")).map(lambda x:(x[0],1))     # 学生每门课程都对应(学生姓名,1),学生有n门课程则有n个(学生姓名,1)
>>> each_res = res.reduceByKey(lambda x,y: x+y)        # 按学生姓名获取每个学生的选课总数
>>> each_res.foreach(print)

执行结果:

......

(5)该系DataBase课程共有多少人选修;

>>> lines = sc.textFile("file:///home/zhc/datasets/data1.txt")
>>> res = lines.map(lambda x:x.split(",")).filter(lambda x:x[1]=="DataBase")
>>> res.count()

执行结果: 

(6)各门课程的平均分是多少;

>>> lines = sc.textFile("file:///home/zhc/datasets/data1.txt")
>>> res = lines.map(lambda x:x.split(",")).map(lambda x:(x[1],(int(x[2]),1)))   # 为每门课程的分数后面新增一列1,表示1个学生选择了该课程。格式如('ComputerNetwork', (44, 1))
>>> temp = res.reduceByKey(lambda x,y:(x[0]+y[0],x[1]+y[1]))    # 按课程名聚合课程总分和选课人数。格式如('ComputerNetwork', (7370, 142))
>>> avg = temp.map(lambda x:(x[0], round(x[1][0]/x[1][1],2)))     # 课程总分/选课人数 = 平均分,并利用round(x,2)保留两位小数
>>> avg.foreach(print)

 执行结果:

(7)使用累加器计算共有多少人选了DataBase这门课。

>>> lines = sc.textFile("file:///home/zhc/datasets/data1.txt")
>>> res = lines.map(lambda x:x.split(",")).filter(lambda x:x[1]=="DataBase")    # 筛选出选了DataBase课程的数据
>>> accum = sc.accumulator(0)        # 定义一个从0开始的累加器accum
>>> res.foreach(lambda x:accum.add(1))        # 遍历res,每扫描一条数据,累加器加1
>>> accum.value           # 输出累加器的最终值

执行结果:

2、编写独立应用程序实现数据去重

        对于两个输入文件A和B,编写Spark独立应用程序,对两个文件进行合并,并剔除其中重复的内容,得到一个新文件C。下面是输入文件和输出文件的一个样例,供参考。

        输入文件A的样例如下:
        20170101    x
        20170102    y
        20170103    x
        20170104    y
        20170105    z
        20170106    z

        输入文件B的样例如下:
        20170101    y
        20170102    y
        20170103    x
        20170104    z
        20170105    y

        根据输入的文件A和B合并得到的输出文件C的样例如下:
        20170101    x
        20170101    y
        20170102    y
        20170103    x
        20170104    y
        20170104    z
        20170105    y
        20170105    z
        20170106    z

在“/home/zhc/mycode/remdup”目录下新建代码文件remdup.py:

# /home/zhc/mycode/remdup/remdup.py
from pyspark import SparkContext
#初始化SparkContext
sc = SparkContext('local','remdup')
#加载两个文件A和B
lines1 = sc.textFile("file:///home/zhc/mycode/remdup/A.txt")
lines2 = sc.textFile("file:///home/zhc/mycode/remdup/B.txt")
#合并两个文件的内容
lines = lines1.union(lines2)
#去重操作
distinct_lines = lines.distinct() 
#排序操作
res = distinct_lines.sortBy(lambda x:x)
#将结果写入result文件中,repartition(1)的作用是让结果合并到一个文件中,不加的话会结果写入到两个文件
res.repartition(1).saveAsTextFile("file:///home/zhc/mycode/remdup/result")

在目录“/home/zhc/mycode/remdup”下执行下面命令执行程序(注意执行程序时请先退出pyspark shell,否则会出现“地址已在使用”的警告)。

[root@bigdata remdup]# python3 remdup.py

在目录“/home/zhc/mycode/remdup/result”下即可得到结果文件part-00000。

[root@bigdata remdup]# cd result
[root@bigdata result]# cat part-00000 

3、编写独立应用程序实现求平均值问题

        每个输入文件表示班级学生某个学科的成绩,每行内容由两个字段组成,第一个是学生名字,第二个是学生的成绩;编写Spark独立应用程序求出所有学生的平均成绩,并输出到一个新文件中。下面是输入文件和输出文件的一个样例,供参考。

        Algorithm成绩:
        小明 92
        小红 87
        小新 82
        小丽 90

        Database成绩:
        小明 95
        小红 81
        小新 89
        小丽 85

        Python成绩:
        小明 82
        小红 83
        小新 94
        小丽 91

        平均成绩如下:
        (小红,83.67)
        (小新,88.33)
        (小明,89.67)
        (小丽,88.67)

在“/home/zhc/mycode/avgscore”目录下新建代码文件avgscore.txt:

# /home/zhc/mycode/avgscore/avgscore.txt
from pyspark import SparkContext
#初始化SparkContext
sc = SparkContext('local',' avgscore')
#加载三个文件Algorithm.txt、Database.txt和Python.txt
lines1 = sc.textFile("file:///home/zhc/mycode/avgscore/Algorithm.txt")
lines2 = sc.textFile("file:///home/zhc/mycode/avgscore/Database.txt")
lines3 = sc.textFile("file:///home/zhc/mycode/avgscore/Python.txt")
#合并三个文件的内容
lines = lines1.union(lines2).union(lines3)
#为每行数据新增一列1,方便后续统计每个学生选修的课程数目。data的数据格式为('小明', (92, 1))
data = lines.map(lambda x:x.split(" ")).map(lambda x:(x[0],(int(x[1]),1)))
#根据key也就是学生姓名合计每门课程的成绩,以及选修的课程数目。res的数据格式为('小明', (269, 3))
res = data.reduceByKey(lambda x,y:(x[0]+y[0],x[1]+y[1]))
#利用总成绩除以选修的课程数来计算每个学生的每门课程的平均分,并利用round(x,2)保留两位小数
result = res.map(lambda x:(x[0],round(x[1][0]/x[1][1],2)))
#将结果写入result文件中,repartition(1)的作用是让结果合并到一个文件中,不加的话会结果写入到三个文件
result.repartition(1).saveAsTextFile("file:///home/zhc/mycode/avgscore/result")

在目录“/home/zhc/mycode/avgscore”下执行下面命令执行程序(注意执行程序时请先退出pyspark shell,否则会出现“地址已在使用”的警告)。

[root@bigdata avgscore]# python3 avgscore.py 

在目录“/home/zhc/mycode/avgscore/result”下即可得到结果文件part-00000。

[root@bigdata avgscore]# cd result
[root@bigdata result]# cat part-00000 

4、三个综合实例

案例一:求Top值

任务描述:某个目录下有若干个文本文件,每个文件里包含了很多数据,每行数据由4个字段的值构成,不同字段之间用逗号隔开,4个字段分别为orderid,userid,payment和productid,要求求出Top N个payment值。

file01.txt:

1,1768,50,155

2,1218, 600,211

3,2239,788,242

4,3101,28,599

5,4899,290,129

6,3110,54,1201

7,4436,259,877

8,2369,7890,27

file02.txt:

100,4287,226,233

101,6562,489,124

102,1124,33,17

103,3267,159,179

104,4569,57,125

105,1438,37,116

[root@bigdata zhc]# cd /mycode/RDD
[root@bigdata RDD]# vi file0.txt
[root@bigdata RDD]# vi TopN.py 
[root@bigdata RDD]# vi file0.txt 
[root@bigdata RDD]# spark-submit TopN.py 

使用vim编辑器编辑“/home/zhc/mycode/RDD/file0.txt”文件:

我这里将file01.txt和file02.txt合并为一个文件了——>file0.txt

1,1768,50,155
2,1218,600,211
3,2239,788,242
4,3101,28,599
5,4899,290,129
6,3110,54,1201
7,4436,259,877
8,2369,7890,27
100,4287,226,233
101,6562,489,124
102,1124,33,17
103,3267,159,179
104,4569,57,125
105,1438,37,116

使用vim编辑器编辑“/home/zhc/mycode/RDD/TopN.py”代码文件:

#/home/zhc/mycode/RDD/TopN.py
from pyspark import SparkConf, SparkContext
# 创建SparkConf对象,设置应用程序名称和部署模式
conf = SparkConf().setMaster("local").setAppName("ReadHBase")
# 创建SparkContext对象
sc = SparkContext(conf = conf)
# 从本地文件系统读取数据
lines= sc.textFile("file:///home/zhc/mycode/RDD/file0.txt")
# 过滤出长度不为0且包含4个逗号的行
result1 = lines.filter(lambda line:(len(line.strip()) > 0) and (len(line.split(","))== 4))
# 提取第三列数据
result2=result1.map(lambda x:x.split(",")[2])
# 将第三列数据转换成键值对(key为数字,value为空串)
result3=result2.map(lambda x:(int(x),""))    
# 对数据进行重新分区,分区数为1
result4=result3.repartition(1)
# 按照键降序排序        
result5=result4.sortByKey(False)
# 取出前5个键 
result6=result5.map(lambda x:x[0])
result7=result6.take(5)
# 打印前5个键         
for a in result7:      print(a)

使用spark-submit提交TopN.py文件,得到结果如下。

案例二:文件排序

任务描述:有多个输入文件,每个文件中的每一行内容均为一个整数。要求读取所有文件中的整数,进行排序后,输出到一个新的文件中,输出的内容个数为每行两个整数,第一个整数为第二个整数的排序位次,第二个整数为原待排序的整数。

输入文件:

file1.txt:

33

37

12

40

file2.txt:

4

16

39

5

file3.txt:

1

45

25

[root@bigdata RDD]# mkdir filesort
[root@bigdata RDD]# cd filesort
[root@bigdata filesort]# vi file1.txt
[root@bigdata filesort]# vi file2.txt
[root@bigdata filesort]# vi file3.txt
[root@bigdata filesort]# cd ..
[root@bigdata RDD]# vi FileSort.py 
[root@bigdata RDD]# spark-submit FileSort.py 

在“/home/zhc/mycode/RDD/filesort”路径下,使用vim编辑器将上面三个文件内容输入。

使用vim编辑器编辑“/home/zhc/mycode/RDD/FileSort.py”文件:

#/home/zhc/mycode/RDD/FileSort.py
from pyspark import SparkConf, SparkContext
# 定义一个全局变量index,用于记录索引值
index=0
# 自定义函数getindex,每调用一次将index加1,并返回新的index值       
def getindex():global indexindex+=1return index
def main():# 创建SparkConf对象,设置应用程序名称和部署模式(本地1核运行)conf = SparkConf().setMaster("local[1]").setAppName("FileSort")   sc = SparkContext(conf = conf)lines= sc.textFile("file:///home/zhc/mycode/RDD/filesort/file*.txt") index = 0# 过滤出长度不为0的行result1 = lines.filter(lambda line:(len(line.strip()) > 0))# 将每行数据转换成整型键值对  result2=result1.map(lambda x:(int(x.strip()),""))   # 对数据进行重新分区,分区数为1result3=result2.repartition(1)# 按照键升序排序          result4=result3.sortByKey(True)# 只保留键        result5=result4.map(lambda x:x[0])# 将数据映射为(index, value)的形式result6=result5.map(lambda x:(getindex(),x)) result6.foreach(print)# 将结果保存到本地文件系统result6.saveAsTextFile("file:///home/zhc/mycode/RDD/filesort/sortresult")
if __name__ == '__main__':main()

使用spark-submit提交FileSort.py文件,得到结果如下。

可以到“/home/zhc/mycode/RDD/filesort/sortresult”目录下查看结果文件part-00000。

[root@bigdata RDD]# cd ./filesort/sortresult
[root@bigdata sortresult]# cat part-00000 

案例三:二次排序

任务描述: 对于一个给定的文件(数据如file4.txt所示),请对数据进行排序,首先根据第1列数据降序排序,如果第1列数据相等,则根据第2列数据降序排序。

输入文件 file4.txt:

5 3
1 6
4 9
8 3
4 7
5 6
3 2

[root@bigdata RDD]# vi file4.txt
[root@bigdata RDD]# vi SecondarySortApp.py 
[root@bigdata RDD]# spark-submit SecondarySortApp.py

在“/home/zhc/mycode/RDD”路径下,使用vim编辑器将上面file4.txt文件内容输入。

使用vim编辑器编辑“/home/zhc/mycode/RDD/SecondarySortApp.py”文件:

#/home/zhc/mycode/RDD/SecondarySortApp.py
# 导入gt函数,用于比较大小
from operator import gt             
from pyspark import SparkContext, SparkConf
# 定义SecondarySortKey类
class SecondarySortKey():def __init__(self, k):self.column1 = k[0]self.column2 = k[1]# 定义__gt__方法,用于比较大小def __gt__(self, other):if other.column1 == self.column1:return gt(self.column2,other.column2)else:return gt(self.column1, other.column1)def main():# 创建SparkConf对象,设置应用程序名称和部署模式(本地1核运行)conf = SparkConf().setAppName('spark_sort').setMaster('local[1]')sc = SparkContext(conf=conf)file="file:///home/zhc/mycode/RDD/file4.txt"rdd1 = sc.textFile(file)# 过滤出长度不为0的行rdd2=rdd1.filter(lambda x:(len(x.strip()) > 0))# 将每行数据转换成带有键值对的元组,键为元组类型rdd3=rdd2.map(lambda x:((int(x.split(" ")[0]),int(x.split(" ")[1])),x))# 将数据中的键转换成SecondarySortKey类型rdd4=rdd3.map(lambda x: (SecondarySortKey(x[0]),x[1]))# 对数据进行按键排序rdd5=rdd4.sortByKey(False)# 只保留值rdd6=rdd5.map(lambda x:x[1])rdd6.foreach(print)if __name__ == '__main__':main()

使用spark-submit提交SecondarySortApp.py文件,得到结果如下。

四、结果分析与实验体会

        在进行RDD编程实验之前,需要掌握Spark的基本概念和RDD的特性,例如惰性计算、分区、依赖关系等。同时需要了解Python等语言的基础知识。在实验过程中,可以通过以下步骤来完成:
(1)创建SparkContext对象,用于连接Spark集群和创建RDD;(2)通过textFile函数读取文件数据,并利用filter等函数进行数据清洗和处理;(3)将数据转换成键值对的形式,再利用map、reduceByKey等函数进行计算和处理;(4)利用sortByKey等函数进行排序操作;(5)最后通过foreach等函数将结果输出。
        在实验过程中,需要注意以下几点:(1)选择合适的算子,例如filter、map、reduceByKey、sortByKey等,以及合适的lambda表达式来进行数据处理和计算。(2)对于大规模数据的处理,需要考虑分区和并行计算,以提高计算效率。(3)需要注意数据类型和格式,确保数据的正确性和一致性。(4)在进行排序操作时,需要利用自定义类来实现二次排序等功能。        
        总之,通过实验可以更加深入地理解Spark的原理和机制,提高数据处理和计算的效率和准确性。同时也能够培养代码编写和调试的能力,提高编程水平。

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

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

相关文章

如何记录函数递归的次数----静态局部变量的使用

记录函数递归的次数,不像是其他普通的函数,只需要一个简单的局部变量作为计数器,每次就好了,函数递归是不断地调用函数,换言之,如果你将一个局部变量定义在函数的内部,那么每次递归都会创建一个这样的变量,每次的值都会初始化,这样也就达不到记录递归次数的目的. 为了解决这个…

极狐GitLab DevSecOps 之容器镜像安全扫描

容器镜像安全 现状 最近某银行遭受供应链攻击的事件传的沸沸扬扬,安全又双叒叕进入了人们的视野。安全确实是一个非常重要,但是又最容易被忽略的话题。但是现在到了一个不得不人人重视安全,人人为安全负责的时代。尤其以现在非常火爆的云原…

【LeetCode: 2276. 统计区间中的整数数目 | 线段树】

🚀 算法题 🚀 🌲 算法刷题专栏 | 面试必备算法 | 面试高频算法 🍀 🌲 越难的东西,越要努力坚持,因为它具有很高的价值,算法就是这样✨ 🌲 作者简介:硕风和炜,…

龙芯loongarch64安装numpy报错“No module named ‘numpy.core._multiarray_umath‘”

前言 在之前编译安装Python3.8的文章中说明了,龙芯仓库的很多包都有问题,安装之后很多无法使用,比如安装numpy后,就会出现“No module named numpy.core._multiarray_umath” 问题复现 配置pip源 vim /etc/pip.conf 复制下面的内容并保存 [global] timeout = 60 index-url…

抖店怎么快速起店?不掺杂汤汤水水,全是干货!

我是电商珠珠 我做抖店也已经有三年的时间了,团队也从原来的几人扩大到了70。对于抖店的玩法已经完全摸透熟通,在做店的同时也会带着学生一起做店,他们经常问的问题就是抖店怎么快速起店。 今天,我就来给大家做个分享。根据我的…

鸿蒙(HarmonyOS)项目方舟框架(ArkUI)之Text文本组件

鸿蒙(HarmonyOS)项目方舟框架(ArkUI)之文本组件 一、操作环境 操作系统: Windows 10 专业版 IDE:DevEco Studio 3.1 SDK:HarmonyOS 3.1 二、文本组件 Text 是显示文本的基础组件之一,它可以包含子组件 Span &…

Python 自动化之收发邮件(一)

imapclient / smtplib 收发邮件 文章目录 imapclient / smtplib 收发邮件前言一、基本内容二、发送邮件1.整体代码 三、获取邮件1.整体代码 总结 前言 简单给大家写个如何用Python进行发邮件和查看邮件教程,希望对各位有所帮助。 一、基本内容 本文主要分为两部分…

选择排序、快速排序和插入排序

1. 选择排序 xuanze_sort.c #include<stdio.h> #include<stdlib.h>//选择排序void xuanze_sort(int arr[],int sz){//正着for(int i0;i<sz;i){//外层循环从第一个数据开始依次作为基准数据for(int j i1;j<sz;j){//int j i1 因为第一个数据作为了基准数据&…

挑战内网安全难题,迅软DSE如何以管控之力助您一臂之力?

随着信息化程度的提高&#xff0c;政企单位面临更加复杂和不可控的内网安全问题。由于缺乏有效技术手段和完善的管理机制&#xff0c;企业内网管理长期处于被动状态&#xff0c;而在发生数据安全事件后的快速处置能力也相对薄弱。 迅软DSE桌面管理系统成为解决方案&#xff0c…

【第2期】Springboot如何快速集成SpringSecurity

简单介绍 本专栏主要结合实战讲解&#xff0c;不过多介绍细节的概念&#xff0c;概念可以通过搜索引擎查找&#xff0c;一搜一大把&#xff0c;切入正题。 本专栏的实战项目是基于SpringbootSpringSecurityRSAJWTVUE的全栈开发项目&#xff0c;每个环节都会专门讲&#xff0c;…

SpringBoot + Vue前后端分离项目实战 || 三:Spring Boot后端与Vue前端连接

系列文章&#xff1a; SpringBoot Vue前后端分离项目实战 || 一&#xff1a;Vue前端设计 SpringBoot Vue前后端分离项目实战 || 二&#xff1a;Spring Boot后端与数据库连接 SpringBoot Vue前后端分离项目实战 || 三&#xff1a;Spring Boot后端与Vue前端连接 SpringBoot V…

力扣200. 岛屿数量(java DFS解法)

Problem: 200. 岛屿数量 文章目录 题目描述思路解题方法复杂度Code 题目描述 思路 该问题可以归纳为一类遍历二维矩阵的题目&#xff0c;此类中的一部分题目可以利用DFS来解决&#xff0c;具体到本题目&#xff1a; 1.我们首先要针对于二维数组上的每一个点&#xff0c;尝试展…