前言
接上次（应该是很久之前写的Scala集合常用方法和函数操作（上）的内容），当时Scala差不多是过了一遍了，但是由于学习 Spark 的过程中，好多方法和函数都是 Scala 中的，而且思路基本和Scala中的集合操作是差不多的，毕竟Spark 的RDD 也可以看做是一个特殊的集合嘛（弹性分布式数据集）。
上次写了一半，但只是学了当时用到的一些操作，一晃快两个多月过去了，最近在学习 Spark Streaming 的过程中又遇到了一些新的集合操作，比如 foldLeft、aggregate等，这次就专门来学习一下。
还有最近情场失意，需要静下心好好学习一段时间了，唉~
Fold、FoldLeft 和 FoldRight
fold(z: A1)((A1,A1)=>A1)：
- fold的意思是折叠，list.fold(z)(_+_)就是从左往右或从右往左与集合内的元素挨个进行累加，因为 fold 其实底层调用的还是 foldLeft
- 要求集合外的参数（z）类型集合内部（list）的参数类型都必须一致。
foldLeft：
- 以 list.foldLeft(z)(_+_)为例
- 参数z 从左向右和集合内的元素挨个进行 op 计算。
- 集合外的参数（z）类型集合内部（list）的参数类型都可以不一致。
foldRight：
- 以list.foldRight(z)(_-_)为例
- 先翻转集合list内的元素
- list内的第 1 个元素先与 z 进行 op（_-_）操作得到 res1
- 再用list内的第 2 个元素与 z 进行 op（_-_）操作得到 res2
- ... 直到list内每个元素都计算完毕
object Test03_Fold { def main(args: Array[String]): Unit = { // 称作集合外的参数 val list = List(1,2,3,4) // fold的底层仍然是调用的 foldLeft // 第一个参数是一个值(称作集合内的参数，必须和集合外的参数类型一致) // 第二个参数是一个函数操作op println(list.fold(5)(_+_)) //15 /** * 5 + 1 = 6 * 6 + 2 = 8 * 8 + 3 = 11 * 11 +4 = 15 */ // 这里的集合内参数类型可以和集合外的参数类型不一致 println(list.foldLeft(5)(_-_)) //执行过程和上面的 fold 一致 // 这里的集合内参数类型可以和集合外的参数类型不一致 println(list.foldRight(5)(_-_)) /** * 翻转集合list => (4,3,2,1) * 4 - 5 = -1 集合第一个值 4 - 初始值 5 = 结果1 * 3 - (-1) = 4 集合第二个值 3 - 结果1 = 结果2 * 2 - 4 = -2 * 1- (-2) = 3 */ } }
MapValues 和 groupBy
mapValues：只适用于 键值对RDD 。只对键值对的值进行操作，相当于（rdd.map{case (k,v): (k,func(v))}）
groupBy：以下面代码为例，对键值对RDD进行操作后得到一个 RDD[String,List[(String,Int)]]
object Test05_MapValues { def main(args: Array[String]): Unit = { val lines = List("hello spark","hello flink","hello flink") val list:Map[String,Int] = lines.flatMap(_.split(" ")) .map((_, 1)) .groupBy(_._1) .mapValues(_.size) for(key <- list.keySet){ println("k= " + key + " v= "+list.getOrElse(key, 0)) } /** * k= hello v= 3 k= spark v= 1 k= flink v= 2 */ } }
Aggregate
第一个参数是我们期望返回的类型，也就是意味着可以返回和我们输入数据不同的数据类型。
第二个参数是一个函数，对于RDD而言，这个函数会把该RDD存放在本地节点的元素合并起来放到累加器，进行计算。
第三个参数也是一个函数，它来对同一RDD存放在不同节点的计算结果的累加器进行两两合并。
import scala.collection.mutable object Test02_Aggregate { def main(args: Array[String]): Unit = { val s = List(1, 2, 3, 4) /** * 需要提供3个参数: * 1.初始值(类型是我们期待返回的类型) * 2.累加器函数：把RDD中的元素合并起来放到累加器进行计算 * 3.合并累加器函数: 由于每个节点在本地计算,所以需要合并不同节点累加器的结果 * (0,0): 作为s的初始值 (类型是我们期待返回的类型) * (s,r): r是s的某个元素(1,2,3,4)中的一个(并行计算,每次取出的值可能是乱序的,但是结果是相同的) * (s._1 + r,s._2 + 1) => (0+1,0+1) => (1,1) * (s._1 + r,s._2 + 1) => (1+2,1+1) => (3,2) * (s._1 + r,s._2 + 1) => (3+3,2+1) => (6,3) * (s._1 + r,s._2 + 1) => (6+4,3+1) => (10,4) */ val r = s.par.aggregate((0, 0))((s, r) =>(s._1 + r, s._2 + 1), (s,r) => (s._1 + r._1, s._2 + r._2)) println(r) // (10,4) val lines = List("hello spark","hello flink","hello flink") var res: mutable.Map[String,Int] = lines.flatMap(_.split(" ")) .aggregate((mutable.Map.empty[String,Int]))((countMap:mutable.Map[String,Int],word)=>{ if(!countMap.contains(word)){ countMap.put(word,1) }else{ countMap.put(word,countMap(word)+1) } countMap }, (map1:mutable.Map[String,Int], map2:mutable.Map[String,Int])=>{ for((word,count)<-map1){ if(!map2.contains(word)){ map2.put(word,1) }else{ map2.put(word,map2(word)+count) } } map2 }) val keys = res.keySet for(key <- keys){ println("k= " + key + " v= "+res.getOrElse(key, 0)) } /** * k= spark v= 1 k= flink v= 2 k= hello v= 3 */ } }
getOrElse
map.getOrElse(key,deault)
用于 Map 数据类型，它的含义是调用 map的get(key)方法获取key对应的值，如果没有返回默认值（也就是第二个参数）。
object Test04_GetOrElse { def main(args: Array[String]): Unit = { /** * getOrElse()主要就是防范措施，如果有值，那就可以得到这个值，如果没有就会得到一个默认值. */ val map: Map[String,Int] = Map("a"->1,"b"->2) println(map.getOrElse("a",0)) // 1 println(map.getOrElse("b",0)) // 2 println(map.getOrElse("c",3)) // 3 } }
