spark实验

“data01.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） 各门课程的平均分是多少； 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 Plain Text

初始化 SparkSession

spark = SparkSession.builder.appName(“GradeAnalysis”).getOrCreate()

读取数据

data_path = “data01.txt” lines = spark.read.text(data_path).rdd

分割数据并转换为 DataFrame

data = lines.map(lambda line: line.value.split(’,’))
.map(lambda x: (x[0], x[1], int(x[2])))
.toDF([“Student”, “Course”, “Score”])

1. 计算该系总共有多少学生

students_count = data.select(“Student”).distinct().count() print(f"学生总数: {students_count}")

2. 计算该系共开设了多少门课程

courses_count = data.select(“Course”).distinct().count() print(f"课程总数: {courses_count}")

3. 计算Tom同学的总成绩平均分

tom_avg_score = data.filter(data.Student == “Tom”).groupBy().agg(avg(“Score”).alias(“AvgScore”)).collect()[0][“AvgScore”] print(f"Tom同学的总成绩平均分: {tom_avg_score}")

4. 计算每名同学的选修的课程门数

students_courses_count = data.groupBy(“Student”).agg(countDistinct(“Course”).alias(“CourseCount”)).show() print(“每名同学的选修的课程门数:”) students_courses_count.show()

5. 计算该系DataBase课程共有多少人选修

db_students_count = data.filter(data.Course == “DataBase”).select(“Student”).distinct().count() print(f"DataBase课程选修人数: {db_students_count}")

6. 计算各门课程的平均分

courses_avg_score = data.groupBy(“Course”).agg(avg(“Score”).alias(“AvgScore”)).show() print(“各门课程的平均分:”) courses_avg_score.show()

停止 SparkSession

spark.stop() 二． from pyspark.sql import SparkSession from pyspark.sql.functions import col def merge_and_deduplicate(fileA, fileB, outputFile):

创建SparkSession

spark = SparkSession.builder
.appName(“MergeAndDeduplicate”)
.getOrCreate()

读取文件A和B，假设它们是以空格分隔的文本文件，并且没有列头

dfA = spark.read.csv(fileA, header=False, inferSchema=True, sep=" “) dfB = spark.read.csv(fileB, header=False, inferSchema=True, sep=” “)

为DataFrame设置列名（这里我们假设只有两列，分别是日期和值）

dfA = dfA.withColumnRenamed("_c0”, “date”).withColumnRenamed("_c1", “value”) dfB = dfB.withColumnRenamed("_c0", “date”).withColumnRenamed("_c1", “value”)

合并两个DataFrame

dfCombined = dfA.unionByName(dfB)

去重（基于两列的组合）

dfDeduplicated = dfCombined.dropDuplicates()

将结果写入新的文件C

dfDeduplicated.write.csv(outputFile, header=True, sep=" “)

停止SparkSession

spark.stop()

调用函数，传入文件路径和输出路径

merge_and_deduplicate(“A.txt”, “B.txt”, “C.txt”) 三． from pyspark.sql import SparkSession from pyspark.sql.functions import avg from pyspark.sql.group import GroupedData def calculate_average_scores(input_files, output_file):

创建SparkSession

spark = SparkSession.builder
.appName(“AverageScoresCalculator”)
.getOrCreate()

读取所有输入文件，假设它们是以空格分隔的文本文件，并且第一列是学生名字，第二列是成绩

由于我们不知道有多少个文件，所以使用reduce来逐个union它们

from functools import reduce dfs = [spark.read.csv(file, header=False, inferSchema=True, sep=” “) for file in input_files] df_combined = reduce(lambda df1, df2: df1.unionByName(df2), dfs)

为DataFrame设置列名

df_combined = df_combined.withColumnRenamed("_c0”, “student_name”).withColumnRenamed("_c1", “score”)

将成绩列转换为数值类型（如果inferSchema没有正确推断的话）

df_combined = df_combined.withColumn(“score”, df_combined[“score”].cast(“double”))

按学生名字分组，并计算平均成绩

grouped = df_combined.groupBy(“student_name”) avg_scores = grouped.agg(avg(“score”).alias(“average_score”))

将结果写入新的文件

avg_scores.write.csv(output_file, header=True, sep=" “)

停止SparkSession

spark.stop()

输入文件列表（这里假设有三个文件，分别代表不同学科的成绩）

input_files = [“1.txt”, “2.txt”, “3.txt”]

输出文件路径

output_file = “average_scores.txt”

调用函数，传入输入文件列表和输出文件路径

calculate_average_scores(input_files, output_file)