SparkCore>RDD基础练习[快速演示]

集群模式启动

启动Spark集群
/export/servers/spark/sbin/start-all.sh

//启动spark-shell 
/export/servers/spark/bin/spark-shell \
--master spark://node01:7077 \
--executor-memory 1g \
--total-executor-cores 2 

//或本地模式启动spark-shell (二选一)
/export/servers/spark/bin/spark-shell 

• WordCount

val res = sc.textFile("hdfs://node01:8020/wordcount/input/words.txt")
.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
//上面的代码不会立即执行,因为都是Transformation转换操作
//下面的代码才会真正的提交并执行,因为是Action动作/行动操作
res.collect//收集打印

• 创建RDD

//parallelize或makeRDD  两种方式都可以创建RDD
val rdd1 = sc.parallelize(List(5,6,4,7,3,8,2,9,1,10))
val rdd2 = sc.makeRDD(List(5,6,4,7,3,8,2,9,1,10))

• 查看该RDD的分区数量

sc.parallelize(List(5,6,4,7,3,8,2,9,1,10)).partitions.length 
//没有指定分区数,默认值是2
sc.parallelize(List(5,6,4,7,3,8,2,9,1,10),3).partitions.length 
//指定了分区数为3
sc.textFile("hdfs://node01:8020/wordcount/input/words.txt").partitions.length 
//2

• RDD分区的数据取决于哪些因素？

RDD分区的原则是使得分区的个数尽量等于集群中的CPU核心(core)数目，这样可以充分利用CPU的计算资源,但是在实际中为了更加充分的压榨CPU的计算资源,会把并行度设置为cpu核数的2~3倍。RDD分区数和启动时指定的核数、调用方法时指定的分区数、如文件本身分区数
有关系

分区原则

• 1.启动的时候指定的CPU核数确定了一个参数值:
  spark.default.parallelism=指定的CPU核数(集群模式最小2)
• 2.对于Scala集合调用parallelize(集合,分区数)方法,
  如果没有指定分区数,就使用spark.default.parallelism,
  如果指定了就使用指定的分区数(不要指定大于spark.default.parallelism)
• 3.对于textFile(文件,分区数) defaultMinPartitions
  如果没有指定分区数sc.defaultMinPartitions=min(defaultParallelism,2)
  如果指定了就使用指定的分区数sc.defaultMinPartitions=指定的分区数

rdd的分区数

• 对于本地文件:
  rdd的分区数 = max(本地file的分片数， sc.defaultMinPartitions)
• 对于HDFS文件:
  rdd的分区数 = max(hdfs文件的block数目， sc.defaultMinPartitions)
  所以如果分配的核数为多个，且从文件中读取数据创建RDD，即使hdfs文件只有1个切片，最后的Spark的RDD的partition数也有可能是2

RDD (API)

http://homepage.cs.latrobe.edu.au/zhe/ZhenHeSparkRDDAPIExamples.html

map

对RDD中的每一个元素进行操作并返回操作的结果

//通过并行化生成rdd
val rdd1 = sc.parallelize(List(5, 6, 4, 7, 3, 8, 2, 9, 1, 10))  
//对rdd1里的每一个元素
rdd1.map(_ * 2).collect  //collect方法表示收集,是action操作

filter

注意:函数中返回True的被留下,返回False的被过滤掉

val rdd2 = sc.parallelize(List(5, 6, 4, 7, 3, 8, 2, 9, 1, 10))
val rdd3 = rdd2.filter(_ >= 10)
rdd3.collect //10

flatmap

对RDD中的每一个元素进行先map再压扁,最后返回操作的结果

val rdd1 = sc.parallelize(Array("a b c", "d e f", "h i j"))
//将rdd1里面的每一个元素先切分再压平
val rdd2 = rdd1.flatMap(_.split(' '))
rdd2.collect
//Array[String] = Array(a, b, c, d, e, f, h, i, j)

sortBy

val rdd1 = sc.parallelize(List(5, 6, 4, 7, 3, 8, 2, 9, 1, 10))
val rdd2 = rdd1.sortBy(x=>x,true) // x=>x 表示按照元素本身进行排序,True表示升序
rdd2.collect //1,2,3,.....
val rdd2 = rdd1.sortBy(x=>x+"",true)//x=>x+""表示按照x的字符串形式排序变成了字符串,结果为字典顺序
rdd2.collect//1,10,2,3...

交集、并集、差集、笛卡尔积

注意类型要一致

val rdd1 = sc.parallelize(List(5, 6, 4, 3))
val rdd2 = sc.parallelize(List(1, 2, 3, 4))
//union不会去重
val rdd3 = rdd1.union(rdd2)
rdd3.collect
//去重
rdd3.distinct.collect
//求交集
val rdd4 = rdd1.intersection(rdd2)
rdd4.collect
//求差集
val rdd5 = rdd1.subtract(rdd2)
rdd5.collect
//笛卡尔积
val rdd1 = sc.parallelize(List("jack", "tom"))//学生
val rdd2 = sc.parallelize(List("java", "python", "scala"))//课程
val rdd3 = rdd1.cartesian(rdd2)//表示所有学生的所有选课情况
rdd3.collect
//Array[(String, String)] = Array((jack,java), (jack,python), (jack,scala), (tom,java), (tom,python), (tom,scala))

join

join(内连接)聚合具有相同key组成的value元组

val rdd1 = sc.parallelize(List(("tom", 1), ("jerry", 2), ("kitty", 3)))
val rdd2 = sc.parallelize(List(("jerry", 9), ("tom", 8), ("shuke", 7), ("tom", 2)))
val rdd3 = rdd1.join(rdd2)
rdd3.collect
//Array[(String, (Int, Int))] = Array((tom,(1,8)), (tom,(1,2)), (jerry,(2,9)))

• 图解
  

val rdd4 = rdd1.leftOuterJoin(rdd2) //左外连接,左边的全留下,右边的满足条件的才留下
rdd4.collect
//Array[(String, (Int, Option[Int]))] = Array((tom,(1,Some(2))), (tom,(1,Some(8))), (jerry,(2,Some(9))), (kitty,(3,None)))

• 图解
  

val rdd5 = rdd1.rightOuterJoin(rdd2)  
rdd5.collect
//Array[(String, (Option[Int], Int))] = Array((tom,(Some(1),2)), (tom,(Some(1),8)), (jerry,(Some(2),9)), (shuke,(None,7)))

val rdd6 = rdd1.union(rdd2)
rdd6.collect
//Array[(String, Int)] = Array((tom,1), (jerry,2), (kitty,3), (jerry,9), (tom,8), (shuke,7), (tom,2))

groupbykey

groupByKey()的功能是，对具有相同键的值进行分组。
比如，对四个键值对(“spark”,1)、(“spark”,2)、(“hadoop”,3)和(“hadoop”,5)，
采用groupByKey()后得到的结果是：(“spark”,(1,2))和(“hadoop”,(3,5))。

//按key进行分组
val rdd6 = sc.parallelize(Array(("tom",1), ("jerry",2), ("kitty",3), ("jerry",9), ("tom",8), ("shuke",7), ("tom",2)))
val rdd7=rdd6.groupByKey
rdd7.collect
//Array[(String, Iterable[Int])] = Array((tom,CompactBuffer(1, 8, 2)), (jerry,CompactBuffer(2, 9)), (shuke,CompactBuffer(7)), (kitty,CompactBuffer(3)))

cogroup[了解]

cogroup是先RDD内部分组,在RDD之间分组

val rdd1 = sc.parallelize(List(("tom", 1), ("tom", 2), ("jerry", 3), ("kitty", 2)))
val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("shuke", 2)))
val rdd3 = rdd1.cogroup(rdd2)
rdd3.collect 
// Array((tom,(CompactBuffer(1, 2),CompactBuffer(1))), (jerry,(CompactBuffer(3),CompactBuffer(2))), (shuke,(CompactBuffer(),CompactBuffer(2))), (kitty,(CompactBuffer(2),CompactBuffer())))

groupBy

根据指定的函数中的规则/key进行分组

val intRdd = sc.parallelize(List(1,2,3,4,5,6))
val result = intRdd.groupBy(x=>{if(x%2 == 0)"even" else "odd"}).collect
//Array[(String, Iterable[Int])] = Array((even,CompactBuffer(2, 4, 6)), (odd,CompactBuffer(1, 3, 5)))

reduce

val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5))
//reduce聚合
val result = rdd1.reduce(_ + _) 
//  第一_ 上次一个运算的结果,第二个_ 这一次进来的元素

★面试题
reduceByKey是Transformation还是Action? --Transformation
reduce是Transformation还是Action? --Action

reducebykey

• 注意reducebykey是转换算子
  reduceByKey(func)的功能是，使用func函数合并具有相同键的值。
  比如，reduceByKey((a,b) => a+b)，有四个键值对(“spark”,1)、(“spark”,2)、(“hadoop”,3)和(“hadoop”,5)
  对具有相同key的键值对进行合并后的结果就是：(“spark”,3)、(“hadoop”,8)。
  可以看出，(a,b) => a+b这个Lamda表达式中，a和b都是指value，
  比如，对于两个具有相同key的键值对(“spark”,1)、(“spark”,2)，a就是1，b就是2。

val rdd1 = sc.parallelize(List(("tom", 1), ("jerry", 3), ("kitty", 2),  ("shuke", 1)))
val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 3), ("shuke", 2), ("kitty", 5)))
val rdd3 = rdd1.union(rdd2) //并集
rdd3.collect
//Array[(String, Int)] = Array((tom,1), (jerry,3), (kitty,2), (shuke,1), (jerry,2), (tom,3), (shuke,2), (kitty,5))
//按key进行聚合
val rdd4 = rdd3.reduceByKey(_ + _)
rdd4.collect
//Array[(String, Int)] = Array((tom,4), (jerry,5), (shuke,3), (kitty,7))

repartition

改变分区数

val rdd1 = sc.parallelize(1 to 10,3) //指定3个分区
//利用repartition改变rdd1分区数
//减少分区
rdd1.repartition(2).partitions.length //新生成的rdd分区数为2
rdd1.partitions.length //3 //注意:原来的rdd分区数不变
//增加分区
rdd1.repartition(4).partitions.length
//减少分区
rdd1.repartition(3).partitions.length
//利用coalesce改变rdd1分区数
//减少分区
rdd1.coalesce(2).partitions.size
rdd1.coalesce(4).partitions.size 

★注意：
repartition可以增加和减少rdd中的分区数，
coalesce默认减少rdd分区数，增加rdd分区数不会生效。
不管增加还是减少分区数原rdd分区数不变,变的是新生成的rdd的分区数

★应用场景:
在把处理结果保存到hdfs上之前可以减少分区数(合并小文件)

sc.textFile("hdfs://node01:8020/wordcount/input/words.txt")
.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
.repartition(1)//在保存到HDFS之前进行重分区为1,那么保存在HDFS上的结果文件只有1个
.saveAsTextFile("hdfs://node01:8020/wordcount/output5")

collect

收集打印信息

val rdd1 = sc.parallelize(List(6,1,2,3,4,5), 2)
rdd1.collect

count

统计个数

//count统计集合中元素的个数
rdd1.count //6

//求RDD中最外层集合里面的元素的个数
val rdd3 = sc.parallelize(List(List("a b c", "a b b"),List("e f g", "a f g"), List("h i j", "a a b")))
rdd3.count //3

distinct

去重

val rdd = sc.parallelize(Array(1,2,3,4,5,5,6,7,8,1,2,3,4), 3)
rdd.distinct.collect 

top

取出最大的前N个

//取出最大的前N个
val rdd1 = sc.parallelize(List(3,6,1,2,4,5))
rdd1.top(2)

take

//按照原来的顺序取前N个
rdd1.take(2) //3 6
//需求:取出最小的2个
rdd1.sortBy(x=>x,true).take(2)

first

按照原来的顺序取前第一个

//按照原来的顺序取前第一个
rdd1.first

keys、values

val rdd1 = sc.parallelize(List("dog", "tiger", "lion", "cat", "panther", "eagle"), 2)
val rdd2 = rdd1.map(x => (x.length, x))
rdd2.collect
//Array[(Int, String)] = Array((3,dog), (5,tiger), (4,lion), (3,cat), (7,panther), (5,eagle))
rdd2.keys.collect
//Array[Int] = Array(3, 5, 4, 3, 7, 5)
rdd2.values.collect
//Array[String] = Array(dog, tiger, lion, cat, panther, eagle)  

mapValues

mapValues表示对RDD中的元素进行操作,Key不变,Value变为操作之后

val rdd1 = sc.parallelize(List((1,10),(2,20),(3,30)))
//_表示每一个value ,key不变,将函数作用于value
val rdd2 = rdd1.mapValues(_*2).collect 
//(1,20),(2,40),(3,60)

collectAsMap

转换成Map

val rdd = sc.parallelize(List(("a", 1), ("b", 2)))
rdd.collectAsMap
//scala.collection.Map[String,Int] = Map(b -> 2, a -> 1)

面试题：foreach和foreachPartition

val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9), 3)
//x是每一个元素  
rdd1.foreach(x => println(x*100)) 
//x是每个分区
rdd1.foreachPartition(x => println(x.reduce(_ + _)))  

• 注意:
  foreach和foreachPartition都是Action操作,但是以上代码在spark-shell中执行看不到输出结果,
  原因是传给foreach和foreachPartition的计算函数是在各个分区执行的,即在集群中的各个Worker上执行的

• 应用场景:
  比如在函数中要将RDD中的元素保存到数据库
  foreach:会将函数作用到RDD中的每一条数据,那么有多少条数据,操作数据库连接的开启关闭就得执行多少次
  foreachPartition:将函数作用到每一个分区,那么每一个分区执行一次数据库连接的开启关闭,有几个分区就会执行数据库连接开启关闭

import org.apache.spark.{SparkConf, SparkContext}

object Test {
  def main(args: Array[String]): Unit = {
    val config = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    val sc = new SparkContext(config)
    //设置日志输出级别
    sc.setLogLevel("WARN")
    val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9), 3)
    //Applies a function f to all elements of this RDD.
    //将函数f应用于此RDD的所有元素
    rdd1.foreach(x => println(x*100))   
//把函数传给各个分区,在分区内循环遍历该分区中的元素 
//x每个元素,即一个一个的数字
    println("==========================")
    //Applies a function f to each partition of this RDD.
    //将函数f应用于此RDD的每个分区
    rdd1.foreachPartition(x => println(x.reduce(_ + _))) 
//把各个分区传递给函数执行 
//x是每个分区
  }
}

面试题：map和mapPartitions

将每一个分区传递给函数

val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9), 3)
//x是每一个分区,y是分区中的元素
rdd1.mapPartitions(x=>x.map(y=>y*2)).collect 

扩展：mapPartitionsWithIndex（同时获取分区号）

功能：
取分区中对应的数据时，还可以将分区的编号取出来，这样就可以知道数据是属于哪个分区的

val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 3)
//该函数的功能是将对应分区中的数据取出来，并且带上分区编号
// 一个index 分区编号
// 一个iter分区内的数据
val func = (index: Int, iter: Iterator[Int]) => {
  iter.map(x => "[partID:" +  index + ", val: " + x + "]")
}

rdd1.mapPartitionsWithIndex(func).collect

//Array[String] = Array(
//[partID:0, val: 1], [partID:0, val: 2], [partID:0, val: 3], 
//[partID:1, val: 4], [partID:1, val: 5], [partID:1, val: 6],
//[partID:2, val: 7], [partID:2, val: 8], [partID:2, val: 9]
//)

扩展：aggregate

val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 3)

//0表示初始值
//第一个_+_,表示区内聚合,第一个_表示历史值,第二个_表示当前值
//第二个_+_,表示区间聚合,第一个_表示历史值,第二个_表示当前值
val result1: Int = rdd1.aggregate(0)(_+_,_+_)  //45  ==> 6 + 15 + 24 = 45
//10表示初始值,每个分区有初始值,区间聚合的时候也有初始值
val result2: Int = rdd1.aggregate(10)(_+_,_+_) //85  ==> 10+ (10+6 + 10+15 + 10+24)=85

扩展：combineByKey

val rdd1 = sc.textFile("hdfs://node01:8020/wordcount/input/words.txt").flatMap(_.split(" ")).map((_, 1))
//Array((hello,1), (me,1), (hello,1), (you,1), (hello,1), (her,1))

//x => x,表示key不变
//(a: Int, b: Int) => a + b:表示区内聚合
//(m: Int, n: Int) => m + n:表示区间聚合
val rdd2 = rdd1.combineByKey(x => x, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)
//val rdd2 = rdd1.combineByKey(x => x, _+_, _+_)//注意这里简写错误,原则:能省则省,不能省则不要偷懒

rdd2.collect
//Array[(String, Int)] = Array((hello,3), (me,1), (you,1), (her,1))

val rddData1: RDD[(String, Float)] = sc.parallelize(
      Array(
        ("班级1", 95f),
        ("班级2", 80f),
        ("班级1", 75f),
        ("班级3", 97f),
        ("班级2", 88f)),
      2)

val rddData2 = rddData1.combineByKey(
      grade => (grade, 1),
      (gc: (Float, Int), grade) => (gc._1 + grade, gc._2 + 1),
      (gc1: (Float, Int), gc2: (Float, Int)) => (gc1._1 + gc2._1, gc1._2 + gc2._2)
    )

val rddData3 = rddData2.map(t => (t._1, t._2._1 / t._2._2))
rddData3.collect

扩展：aggregateByKey

val pairRDD = sc.parallelize(List( ("cat",2), ("cat", 5), ("mouse", 4),("cat", 12), ("dog", 12), ("mouse", 2)), 2)

def func(index: Int, iter: Iterator[(String, Int)]) : Iterator[String] = {
  iter.map(x => "[partID:" +  index + ", val: " + x + "]")
}
pairRDD.mapPartitionsWithIndex(func).collect
//Array(
[partID:0, val: (cat,2)], [partID:0, val: (cat,5)], [partID:0, val: (mouse,4)], 
[partID:1, val: (cat,12)], [partID:1, val: (dog,12)], [partID:1, val: (mouse,2)]
)

pairRDD.aggregateByKey(0)(math.max(_, _), _ + _).collect 
// Array[(String, Int)] = Array((dog,12), (cat,17), (mouse,6))
//100表示区内初始值,区间聚合没有
pairRDD.aggregateByKey(100)(math.max(_, _), _ + _).collect
//Array[(String, Int)] = Array((dog,100), (cat,200), (mouse,200))  

pairRDD.aggregateByKey(5)(math.max(_, _), _ + _).collect
//Array[(String, Int)] = Array((dog,12), (cat,17), (mouse,10))

pairRDD.aggregateByKey(10)(math.max(_, _), _ + _).collect
//Array[(String, Int)] = Array((dog,12), (cat,22), (mouse,20))

val rddData1 = sc.parallelize(
      Array(
        ("用户1", "接口1"),
        ("用户2", "接口1"),
        ("用户1", "接口1"),
        ("用户1", "接口2"),
        ("用户2", "接口3")),
      2)
val rddData2 = rddData1.aggregateByKey(collection.mutable.Set[String]())(
      (urlSet, url) => urlSet += url,
      (urlSet1, urlSet2) => urlSet1 ++= urlSet2)
rddData2.collect

小练习

• 需求
  给定一个键值对RDD

val rdd = sc.parallelize(Array(("spark",2),("hadoop",6),("hadoop",4),("spark",6)))

key表示图书名称，
value表示某天图书销量，
请计算每个键对应的平均值，也就是计算每种图书的每天平均销量。
最终结果:(“spark”,4),(“hadoop”,5)

val rdd1 = rdd.groupByKey 
rdd1.collect
//Array((spark,CompactBuffer(6, 2)), (hadoop,CompactBuffer(4, 6)))
val rdd2 = rdd1.mapValues(v => v.sum / v.size) 
rdd2.collect

• 答案

val rdd = sc.parallelize(Array(("spark",2),("hadoop",6),("hadoop",4),("spark",6)))
val rdd2 = rdd.groupByKey()
rdd2.collect
//Array[(String, Iterable[Int])] = Array((spark,CompactBuffer(2, 6)), (hadoop,CompactBuffer(6, 4)))

val rdd3 = rdd2.map(t=>(t._1,t._2.sum /t._2.size))
rdd3.collect
//Array[(String, Int)] = Array((spark,4), (hadoop,5))

总结

分类

• RDD的算子分为两类，一类是Transformation转换操作，一类是Action动作操作

如何区分Transformation和Action

• 返回值是RDD的为Transformation转换操作,延迟执行/懒执行/惰性执行
• 返回值不是RDD(如Unit、Array、Int)的为Action动作操作

面试题：

• 1.Transformation操作的API有哪些？ --map/flatMap/filter…
• 2.Action操作的API有哪些？ --collect/reduce/saveAsTextFile…
• 3.reduceByKey是Transformation还是Action？ --Transformation
• 4.reduce是Transformation还是Action？ – Action
• 5.foreach和foreachPartition的区别? foreach作用于每个元素,foreachPartition作用于每个分区

注意:

• RDD不实际存储真正要计算的数据，而只是记录了RDD的转换关系(调用了什么方法，传入什么函数，依赖哪些RDD，分区器是什么，数量块来源机器列表)
  RDD中的所有转换操作都是延迟执行(懒执行)的，也就是说并不会直接计算。只有当发生Action操作的时候，这些转换才会真正运行。