寒假学习记录16

寒假学习记录16

实验 4  RDD 编程初级实践

一、实验目的

(1)熟悉 Spark 的 RDD 基本操作及键值对操作;

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

二、实验平台

操作系统:Ubuntu16.04 Spark 版本:2.1.0

三、实验内容和要求

1.spark-shell 交互式编程

请到本教程官网的“下载专区”的“数据集”中下载 chapter5-data1.txt,该数据集包含 了某大学计算机系的成绩,数据格式如下所示:

请根据给定的实验数据,在 spark-shell 中通过编程来计算以下内容:

(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

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.spark.HashPartitioner

object RemDup
{
    def main(args:Array[String])
    {
        val conf = new SparkConf().setAppName("RemDup")
        val sc = new SparkContext(conf)
        val dataFile = "file:///usr/local/spark/sparksqldata/data4_2"
        val data = sc.textFile(dataFile,2)
        val res = data.filter(_.trim().length>0).map(line=>(line.trim,"")).partitionBy(new HashPartitioner(1)).groupByKey().sortByKey().keys
        res.saveAsTextFile("result")
    }
}

  

  

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)

package my.scala
import org.apache.spark.{SparkConf, SparkContext}
object pingjunzhi {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("reduce")
    val sc = new SparkContext(conf)
    sc.setLogLevel("ERROR")
   
val fourth = sc.textFile("file:///usr/local/spark/sparksqldata/data4_3 ")
  
val res = fourth.filter(_.trim().length>0).map(line=>(line.split("	")(0).trim(),line.split("	")(1).trim().toInt)).groupByKey().map(x => {
   var num = 0.0
   var sum = 0
   for(i <- x._2){
    sum = sum + i
    num = num +1
   }
   val avg = sum/num
   val format = f"$avg%1.2f".toDouble
   (x._1,format)
 }).collect.foreach(x => println(x._1+"	"+x._2))
  }
}

  

原文地址:https://www.cnblogs.com/zhoulonghai/p/12292383.html