pyspark实验

2023-03-31 12:57| 来源: 网络整理| 查看: 265

（1）该系一共多少名学生。

line=sc.textFile("file:///opt/") rdd=line.map(lambda x:x.split(",")).map(lambda x:x[0]) sum=rdd.distinct() sum.count()

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

line=sc.textFile("file:///opt/") rdd=line.map(lambda x:x.split(",")).map(lambda x:x[1]) sum=rdd.distinct() sum.count()

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

#导入 line=sc.textFile("file:///opt/") rdd2=line.map(lambda x:x.split(",")).filter(lambda x:x[0]=="Tom") rdd2.foreach(print)

若出现如下错误则加上

from __future__ import print_function

在这里插入图片描述

#统计 sc=rdd2.map(lambda x:int(x[2])) sum=rdd2.count() nums=sc.reduce(lambda x,y:x+y) avgs=nums/sum print(avgs)

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

line=sc.textFile("file:///opt/") rdd3=line.map(lambda x:x.split(",")).map(lambda x:(x[0],1)) sums2=rdd3.reduceByKey(lambda x,y:x+y) sums2.foreach(print)

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

line=sc.textFile("file:///opt/") rdd5=line.map(lambda x:x.split(",")).filter(lambda x:x[1]=="DataBase") rdd5.count()

（6)各门课程的平均分是多少。

line=sc.textFile("file:///opt/") rdd6=line.map(lambda x:x.split(",")).map(lambda x:(x[1],(int(x[2]),1))) sum6=rdd6.reduceByKey(lambda x,y:(x[0]+y[0],x[1]+y[1])) avg2=sum6.map(lambda x:(x[0],round(x[1][0]/x[1][1],2))) avg2.foreach(print)

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

line=sc.textFile("file:///opt/") rdd7=line.map(lambda x:x.split(",")).filter(lambda x:x[1]=="DataBase") sum7=sc.accumulator(0) rdd7.foreach(lambda x:sum7.add(1)) sum7.value

去重

#!/user/bin/env python3 from pyspark import SparkConf,SparkContext conf=SparkConf().setMaster("local").setAppName("CPayment") sc=SparkContext(conf=conf) line1=sc.textFile("file:///opt/.../A.txt") line2=sc.textFile("file:///opt/.../B.txt") line3=line1.union(line2) distinct_line3=line3.distinct() res=distinct_line3.sortBy(lambda x:x) res.repartition(1).saveAsTextFile("file:///opt.../result")

求平均值

#!/user/bin/env python3 from pyspark import SparkConf,SparkContext conf=SparkConf().setMaster("local").setAppName("AVGPartemnt") sc=SparkContext(conf=conf) line1=sc.textFile("file:///opt../Algorithm.txt") line1=sc.textFile("file:///opt../DataBase.txt") line1=sc.textFile("file:///opt../Python.txt") lines=line1.union(line2).union(line3) database=lines.map(lambda x:x.split(",")).map(lambda x:(x[0],(int(x[1]).1))) #根据key也就是学生姓名合计每门课的成绩，以及选修课的成绩 res=database.reduceByKey(lambda x,y:(x[0]+y[0],x[1]+y[1])) #利用总成绩除以选修的成绩计算平均值，并保留round保留两位小数 result=res.map(lambda x:x[0],round(x[1][0]/x[1][1],2))) result.repartition(1).saveAsTextFile("file:///opt.../avg")

【本文地址】

公司简介

联系我们