掌握RDD算子2

扁平映射算子案例

任务1、统计不规则二维列表元素个数

方法一、利用Scala来实现

• 在net.xxr.rdd.day01包里创建Example02单例对象

package net.xxr.rdd.day01import org.apache.spark.{SparkConf, SparkContext}/*** 功能：利用Scala统计不规则二维列表元素个数*/
object Example02 {def main(args: Array[String]): Unit = {// 创建不规则二维列表val mat = List(List(7, 8, 1, 5),List(10, 4, 9),List(7, 2, 8, 1, 4),List(21, 4, 7, -4))// 输出二维列表println(mat)// 将二维列表扁平化为一维列表val arr = mat.flatten// 输出一维列表println(arr)// 输出元素个数println("元素个数：" + arr.size)}
}

方法二、利用Spark RDD来实现

• 在net.xxr.rdd.day01包里创建Example03单例对象

package net.xxr.rdd.day01import org.apache.spark.{SparkConf, SparkContext}/*** 功能：利用RDD统计不规则二维列表元素个数*/
object Example03 {def main(args: Array[String]): Unit = {// 创建Spark配置对象val conf = new SparkConf().setAppName("PrintDiamond") // 设置应用名称.setMaster("local[*]") // 设置主节点位置（本地调试）// 基于Spark配置对象创建Spark容器val sc = new SparkContext(conf)// 创建不规则二维列表val mat = List(List(7, 8, 1, 5),List(10, 4, 9),List(7, 2, 8, 1, 4),List(21, 4, 7, -4))// 基于二维列表创建rdd1val rdd1 = sc.makeRDD(mat)// 输出rdd1rdd1.collect.foreach(x => print(x + " "))println()// 进行扁平化映射val rdd2 = rdd1.flatMap(x => x.toString.substring(5, x.toString.length - 1).split(", "))// 输出rdd2rdd2.collect.foreach(x => print(x + " "))println()// 输出元素个数println("元素个数：" + rdd2.count)}
}

• 扁平化映射可以简化
  
  

按键归约算子案例

任务1、在Spark Shell里计算学生总分

• 创建成绩列表scores，基于成绩列表创建rdd1，对rdd1按键归约得到rdd2，然后查看rdd2内容

val scores = List((“张钦林”, 78), (“张钦林”, 90), (“张钦林”, 76),
(“陈燕文”, 95), (“陈燕文”, 88), (“陈燕文”, 98),
(“卢志刚”, 78), (“卢志刚”, 80), (“卢志刚”, 60))
val rdd1 = sc.makeRDD(scores)
val rdd2 = rdd1.reduceByKey((agg, cur) => agg + cur)
rdd2.collect.foreach(println)

• 可以采用神奇的占位符
  

任务2、在IDEA里计算学生总分

第一种方式：读取二元组成绩列表

• 在net.xxr.rdd.day02包里创建CalculateScoreSum01单例对象

package net.xxr.rdd.day02import org.apache.spark.{SparkConf, SparkContext}/*** 功能：计算总分*/
object CalculateScoreSum01 {def main(args: Array[String]): Unit = {// 创建Spark配置对象val conf = new SparkConf().setAppName("PrintDiamond") // 设置应用名称.setMaster("local[*]") // 设置主节点位置（本地调试）// 基于Spark配置对象创建Spark容器val sc = new SparkContext(conf)val scores = List(("张钦林", 78), ("张钦林", 90), ("张钦林", 76),("陈燕文", 95), ("陈燕文", 88), ("陈燕文", 98),("卢志刚", 78), ("卢志刚", 80), ("卢志刚", 60))// 基于二元组成绩列表创建RDDval rdd1 = sc.makeRDD(scores)// 对成绩RDD进行按键归约处理val rdd2 = rdd1.reduceByKey(_ + _)// 输出归约处理结果rdd2.collect.foreach(println)}
}

第二种方式：读取四元组成绩列表

• 在net.xxr.rdd.day02包里创建CalculateScoreSum02单例对象

package net.xxr.rdd.day02import org.apache.spark.{SparkConf, SparkContext}import scala.collection.mutable.ListBuffer/*** 功能：计算总分*/
object CalculateScoreSum02 {def main(args: Array[String]): Unit = {// 创建Spark配置对象val conf = new SparkConf().setAppName("PrintDiamond") // 设置应用名称.setMaster("local[*]") // 设置主节点位置（本地调试）// 基于Spark配置对象创建Spark容器val sc = new SparkContext(conf)// 创建四元组成绩列表val scores = List(("张钦林", 78, 90, 76),("陈燕文", 95, 88, 98),("卢志刚", 78, 80, 60))// 将四元组成绩列表转化成二元组成绩列表val newScores = new ListBuffer[(String, Int)]()// 通过遍历算子遍历四元组成绩列表scores.foreach(score => {newScores.append(Tuple2(score._1, score._2))newScores.append(Tuple2(score._1, score._3))newScores.append(Tuple2(score._1, score._4))})// 基于二元组成绩列表创建RDDval rdd1 = sc.makeRDD(newScores)// 对成绩RDD进行按键归约处理val rdd2 = rdd1.reduceByKey(_ + _)// 输出归约处理结果rdd2.collect.foreach(println)}
}

第三种情况：读取HDFS上的成绩文件

• 将成绩文件上传到HDFS的/input目录

hdfs dfs -mkdir /input
hdfs dfs -put scores.txt /input
hdfs dfs -cat /input/scores.txt

• 在net.xxr.rdd.day02包里创建CalculateScoreSum03单例对象

package net.xxr.rdd.day02import org.apache.spark.{SparkConf, SparkContext}import scala.collection.mutable.ListBuffer/*** 功能：计算总分*/
object CalculateScoreSum03 {def main(args: Array[String]): Unit = {// 创建Spark配置对象val conf = new SparkConf().setAppName("CalculateScoreSum").setMaster("local[*]")// 基于配置创建Spark上下文val sc = new SparkContext(conf)// 读取成绩文件，生成RDDval lines = sc.textFile("hdfs://master:9000/input/scores.txt")// 定义二元组成绩列表val scores = new ListBuffer[(String, Int)]()// 遍历lines，填充二元组成绩列表lines.collect.foreach(line => {val fields = line.split(" ")scores += Tuple2(fields(0), fields(1).toInt)scores += Tuple2(fields(0), fields(2).toInt)scores += Tuple2(fields(0), fields(3).toInt)})// 基于二元组成绩列表创建RDDval rdd1 = sc.makeRDD(scores)// 对成绩RDD进行按键归约处理val rdd2 = rdd1.reduceByKey((x, y) => x + y)// 输出归约处理结果rdd2.collect.foreach(println)}
}

• 在Spark Shell里完成同样的任务

import scala.collection.mutable.ListBuffer
val lines = sc.textFile("hdfs://master:9000/input/scores.txt")
val scores = new ListBuffer[(String, Int)]()
lines.collect.foreach(line => {
val fields = line.split(" ")
scores.append(Tuple2(fields(0), fields(1).toInt))
scores.append(Tuple2(fields(0), fields(2).toInt))
scores.append(Tuple2(fields(0), fields(3).toInt))
})
val rdd1 = sc.makeRDD(scores)
val rdd2 = rdd1.reduceByKey(_ + _)
rdd2.collect.foreach(println)

• 修改程序，将计算结果写入HDFS文件