本文共 4431 字，大约阅读时间需要 14 分钟。

1、实验目的

（1）熟悉Spark的RDD基本操作及键值对操作；

（2）熟悉使用RDD编程解决实际具体问题的方法。

2、实验平台

操作系统：Ubuntu16.04

Spark版本：2.4.0 Python版本：3.4.3

3、数据来源

4、实验内容和要求

4.1、pyspark交互式编程（实验描述）

本作业提供分析数据data.txt，该数据集包含了某大学计算机系的成绩，数据格式如下所示：

Tom,DataBase,80Tom,Algorithm,50Tom,DataStructure,60Jim,DataBase,90Jim,Algorithm,60Jim,DataStructure,80……

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

（1）该系总共有多少学生；

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

4.2、编写独立应用程序实现数据去重（实验描述）

对于两个输入文件A和B，编写Spark独立应用程序，对两个文件进行合并，并剔除其中重复的内容，得到一个新文件C。本文给出门课的成绩（A.txt、B.txt）下面是输入文件和输出文件的一个样例，供参考。

输入文件A的样例如下：

20200101 x

20200102 y

20200103 x

20200104 y

20200105 z

20200106 z

输入文件B的样例如下：

20200101 y

20200102 y

20200103 x

20200104 z

20200105 y

根据输入的文件A和B合并得到的输出文件C的样例如下：

20200101 x

20200101 y

20200102 y

20200103 x

20200104 y

20200104 z

20200105 y

20200105 z

20200106 z

4.3、编写独立应用程序实现求平均值问题（实验描述）

每个输入文件表示班级学生某个学科的成绩，每行内容由两个字段组成，第一个是学生名字，第二个是学生的成绩；编写Spark独立应用程序求出所有学生的平均成绩，并输出到一个新文件中。本文给出门课的成绩（Algorithm.txt、Database.txt、Python.txt），下面是输入文件和输出文件的一个样例，供参考。

Algorithm成绩：

小明 92

小红 87

小新 82

小丽 90

Database成绩：

小明 95

小红 81

小新 89

小丽 85

Python成绩：

小明 82

小红 83

小新 94

小丽 91

平均成绩如下：

(小红,83.67)

(小新,88.33)

(小明,89.67)

(小丽,88.67)

5、数据处理过程描述

5.1pyspark交互式编程（实验操作及结果）

1、数据导入

在/usr/local/spark目录下创建一个sparksqldata文件夹，并将data.txt放进去

2、根据给定的实验数据，在pyspark中通过编程来计算以下内容

（1）该系总共有多少学生；

关键代码

lines= sc.textFile("file:///usr/local/spark/sparksqldata/data.txt")res = lines.map(lambda x:x.split(",")).map(lambda x:x[0])sum = res.distinct()sum.count()

实验结果

（2）该系共开设了多少门课程；

关键代码

lines= sc.textFile("file:///usr/local/spark/sparksqldata/data.txt")res = lines.map(lambda x:x.split(",")).map(lambda x:x[1])sum = res.distinct()sum.count()

实验结果

（3）Tom同学的总成绩平均分是多少；

关键代码

lines= sc.textFile("file:///usr/local/spark/sparksqldata/data.txt")res = lines.map(lambda x:x.split(",")).filter(lambda x:x[0]=='Tom')res.foreach(print)score=res.map(lambda x:int(x[2]))sum_score=score.reduce(lambda x,y:x+y)num=res.count()avg = sum_score/numprint(avg)

实验结果

（4）求每名同学的选修的课程门数；

关键代码

lines= sc.textFile("file:///usr/local/spark/sparksqldata/data.txt")res = lines.map(lambda x:x.split(",")).map(lambda x:(x[0],1))each_res = res.reduceByKey(lambda x,y:x+y)each_res.foreach(print)

实验结果

（5）该系DataBase课程共有多少人选修

关键代码

lines= sc.textFile("file:///usr/local/spark/sparksqldata/data.txt")res=lines.map(lambda x:x.split(",")).filter(lambda x:x[1]=='DataBase')res.count()

实验结果

（6）各门课程的平均分是多少；

关键代码

lines= sc.textFile("file:///usr/local/spark/sparksqldata/data.txt")res = lines.map(lambda x:x.split(",")).map(lambda x:(x[1],(int(x[2]),1)))temp = res.reduceByKey(lambda x,y:(x[0]+y[0],x[1]+y[1]))avg = temp.map(lambda x:(x[0],round(x[1][0]/x[1][1],2)))avg.foreach(print)

实验结果

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

关键代码

lines= sc.textFile("file:///usr/local/spark/sparksqldata/data.txt")res = lines.map(lambda x:x.split(",")).filter(lambda x:x[1]=="DataBase")accum = sc.accumulator(0)res.foreach(lambda x:accum.add(1))accum.value

实验结果

5.2、编写独立应用程序实现数据去重（实验操作及结果）

1、数据导入

将A.txt B.txt 导入/usr/local/spark/mycode/remdup目录下

2、在当前目录下新建一个remdup.py文件 3、在remdup.py里进行编写程序

from pyspark import SparkContextsc = SparkContext('local','remdup')lines1 = sc.textFile("file:///usr/local/spark/mycode/remdup/A.txt")lines2 = sc.textFile("file:///usr/local/spark/mycode/remdup/B.txt")lines = lines1.union(lines2)distinct_lines = lines.distinct()res = distinct_lines.sortBy(lambda x:x)res.repartition(1).saveAsTextFile("file:///usr/local/spark/mycode/remdup/C.txt")

4、运行程序

5、查看结果

进入/usr/local/spark/mycode/remdup目录下

5.3编写独立应用程序实现求平均值问题（实验操作及结果）

1、数据导入

将Algorithm.txt、Database.txt、Python.txt 导入/usr/local/spark/mycode/avgscore目录下

2.在当前目录下新建一个avgscore.py文件

3、在avgscore.py里进行编写程序

from pyspark import SparkContextsc= SparkContext('local','avgscore')lines1 = sc.textFile("file:///usr/local/spark/mycode/avgscore/Algorithm.txt")lines2 = sc.textFile("file:///usr/local/spark/mycode/avgscore/Database.txt")lines3 = sc.textFile("file:///usr/local/spark/mycode/avgscore/Python.txt")lines = lines1.union(lines2).union(lines3)distinct_lines = lines.distinct()lines4 = distinct_lines.sortBy(lambda x:x).filter(bool)data = lines4.map(lambda x:x.split(" ")).map(lambda x:(x[0],(int(x[1]),1)))res = data.reduceByKey(lambda x,y:(x[0]+y[0],x[1]+y[1]))result = res.map(lambda x:(x[0],round(x[1][0]/x[1][1],2)))result.repartition(1).saveAsTextFile("file:///usr/local/spark/mycode/avgscore/result")

4、运行程序

5、查看结果

进入/usr/local/spark/mycode/avgscore目录下

转载地址：http://iixvi.baihongyu.com/