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

发布时间：2023年12月18日

目录

一、目的与要求

二、实验内容

三、实验步骤

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 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的原理和机制，提高数据处理和计算的效率和准确性。同时也能够培养代码编写和调试的能力，提高编程水平。

文章来源:https://blog.csdn.net/Morse_Chen/article/details/135049506
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
最新文章