一、目的与要求
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 index
index+=1
return 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()),""))
# 对数据进行重新分区，分区数为1
result3=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的原理和机制，提高数据处理和计算的效率和准确性。同时也能够培养代码编写和调试的能力，提高编程水平。
版权归原作者 Francek Chen 所有， 如有侵权，请联系我们删除。