spark-shell的Scala的一些方法详解

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了spark-shell的Scala的一些方法详解，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含5615字，纯文字阅读大概需要9分钟。

内容图文

Tom,DataBase,80

Tom,Algorithm,50

Tom,DataStructure,60

Jim,DataBase,90

Jim,Algorithm,60

Jim,DataStructure,80

.......

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

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

val lines = sc.textFile("file:///usr/local/spark/sparksqldata/Data01.txt")
val par = lines.map(row=>row.split(",")(0))
val distinct_par = par.distinct() //去重操作
distinct_par.count //取得总数

　　

答案为：265 人

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

val lines = sc.textFile("file:///usr/local/spark/sparksqldata/Data01.txt")
val par = lines.map(row=>row.split(",")(1))//根据，切分的每行数据的第二列进行map
val distinct_par = par.distinct()//去重
distinct_par.count//取总数

　　答案为 8 门

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

val lines = sc.textFile("file:///usr/local/spark/sparksqldata/Data01.txt")
val pare = lines.filter(row=>row.split(",")(0)=="Tom")
pare.foreach(println)
Tom,DataBase,26
Tom,Algorithm,12
Tom,OperatingSystem,16
Tom,Python,40
Tom,Software,60
pare.map(row=>(row.split(",")(0),row.split(",")(2).toInt))
.mapValues(x=>(x,1)).//mapValues是对值的操作,不操作key使数据变成(Tom,（26,1）)
reduceByKey((x,y) => (x._1+y._1,x._2 + y._2))//接着需要按key进行reduce，让key合并当将Tom进行reduce后 这里的(x,y) 表示的是(26,1)(12,1)
.mapValues(x => (x._1 / x._2))//接着要对value进行操作，用mapValues()就行啦
.collect()
//res9: Array[(String, Int)] = Array((Tom,30))

　　Tom 同学的平均分为 30 分

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

val lines = sc.textFile("file:///usr/local/spark/sparksqldata/Data01.txt")
val pare = lines.map(row=>(row.split(",")(0),row.split(",")(1)))
pare.mapValues(x => (x,1))//数据变为(Tom,(DataBase,1)),(Tom,(Algorithm,1)),(Tom,(OperatingSystem,1)),(Tom,(Python,1)),(Tom,(Software,1))

.reduceByKey((x,y) => (" ",x._2 + y._2))//数据变为(Tom,( ,5))

.mapValues(x =>x._2)//数据变为(Tom, 5)

.foreach(println)

　　答案共 265 行

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

val lines = sc.textFile("file:///usr/local/spark/sparksqldata/Data01.txt")
val pare = lines.filter(row=>row.split(",")(1)=="DataBase")filter方法允许你提供一个判断条件（函数），来过滤集合元素
pare.count
res1: Long = 126

　　答案为 126 人

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

val lines = sc.textFile("file:///usr/local/spark/sparksqldata/Data01.txt")
val pare = lines.map(row=>(row.split(",")(1),row.split(",")(2).toInt))
pare.mapValues(x=>(x,1)).reduceByKey((x,y) => (x._1+y._1,x._2 + y._2)).mapValues(x => (x._1 / x._2)).collect()
res0: Array[(String, Int)] = Array((Python,57), (OperatingSystem,54), (CLanguage,50),
(Software,50), (Algorithm,48), (DataStructure,47), (DataBase,50), (ComputerNetwork,51))

　　答案为： (CLanguage,50) (Python,57) (Software,50) (OperatingSystem,54) (Algorithm,48) (DataStructure,47) (DataBase,50) (ComputerNetwork,51)

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

val lines = sc.textFile("file:///usr/local/spark/sparksqldata/Data01.txt")
val pare = lines.filter(row=>row.split(",")(1)=="DataBase").map(row=>(row.split(",")(1),1))
val accum = sc.longAccumulator("My Accumulator")//累加器函数Accumulator
pare.values.foreach(x => accum.add(x))
accum.value
res19: Long = 126

　　答案：共有 126 人

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

　　eclipse代码

package my.scala
import org.apache.spark.{SparkConf, SparkContext}
object case2 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("reduce")
    val sc = new SparkContext(conf)
    sc.setLogLevel("ERROR")
    //获取数据
    val two = sc.textFile("hdfs://192.168.85.128:9000/quchong")
    two.filter(_.trim().length>0) //需要有空格。
        .map(line=>(line.trim,""))//全部值当key，(key value,"")
          .groupByKey()//groupByKey,过滤重复的key value ，发送到总机器上汇总
              .sortByKey() //按key value的自然顺序排序
                  .keys.collect().foreach(println) //所有的keys变成数组再输出
    //第二种有风险
    two.filter(_.trim().length>0)
          .map(line=>(line.trim,"1"))
            .distinct()
                .reduceByKey(_+_)
                    .sortByKey()
                        .foreach(println)

    //reduceByKey,在本机suffle后,再发送一个总map，发送到一个总机器上汇总，（汇总要压力小）
    //groupByKey,发送本机所有的map,在一个机器上汇总（汇总压力大）
    //如果数据在不同的机器上，则会出现先重复数据，distinct，reduceBykey，只是在本机上去重，谨慎一点的话，在reduceByKey后面需要加多一个distinct

  }
}

　　

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("hdfs://192.168.85.128:9000/pingjunzhi")
 
val res = fourth.filter(_.trim().length>0).map(line=>(line.split("\t")(0).trim(),line.split("\t")(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+"\t"+x._2))
  }
}

　　

 

内容总结

以上是互联网集市为您收集整理的spark-shell的Scala的一些方法详解全部内容，希望文章能够帮你解决spark-shell的Scala的一些方法详解所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

---