创建RDD

1.由外部存储系统的数据集创建，包括本地的文件系统，还有所有Hadoop支持的数据集，比如HDFS、Cassandra、HBase等

val rdd1 = sc.textFile("hdfs://node-01:9000/wordcount/input/words.txt")

2.通过已有的RDD经过算子转换生成新的RDD

val rdd2=rdd1.flatMap(_.split(" "))

3.由一个已经存在的Scala集合创建

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

或者

val rdd4 = sc.makeRDD(List(1,2,3,4,5,6,7,8))

makeRDD方法底层调用了parallelize方法

RDD的方法/算子分类

●分类

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

Transformation转换操作:返回一个新的RDD

Action动作操作:返回值不是RDD(无返回值或返回其他的)

●注意:

RDD不实际存储真正要计算的数据，而是记录了数据的位置在哪里,RDD的转换关系(调用了什么方法，传入什么函数)

RDD中的所有转换都是惰性求值/延迟执行的，也就是说并不会直接计算。

只有当发生一个要求返回结果给Driver的Action动作时，这些转换才会真正运行。

之所以使用惰性求值/延迟执行，是因为这样可以在Action时对RDD操作形成的DAG有向无环图进行Stage的划分和并行优化

这种设计让Spark更加有效率地运行。

●总结：如何区分

返回值是RDD的为Transformation转换操作

返回值不是RDD(如Unit、Array)的为Action动作操作

Transformation算子(转换)

Action算子(行动/动作)

统计操作(属于Action)

基础练习[快速演示]

1.准备工作

●集群模式启动

启动Spark集群

/opt/soft/spark/sbin/start-all.sh

启动spark-shell

/opt/soft/spark/bin/spark-shell \

--master spark://node-01:7077,node-02:7077 \

--executor-memory 1g \

--total-executor-cores 2

●或本地模式启动

/opt/soft/spark/bin/spark-shell

2.API演示

2.1 WordCount

val res = sc.textFile("hdfs://node-01:9000/wordcount/input/words.txt").flatMap(x=>x.split(" ")).map((_,1)).reduceByKey(_+_)

//上面的代码不会立即执行,因为都是Transformation转换操作

//下面的代码才会真正的提交并执行,因为是Action动作/行动操作

res.collect

2.2创建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))

2.3查看该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://node-01:9000/wordcount/input/words.txt").partitions.length //2

Hadoop的分区:目的并发读写

hive的分区、分桶:hive分区就是分文件夹,目的是提高查询效率,hive的分桶就是分文件,目的提高查询效率

Kafak的分区:目的并发读写

HBase的分区(Region):数据会按照Rowkey的范围放到不同的Region,目的也是提高效率

注意:HBase写比读快,底层是LSM树决定的

Spark的分区:目的是为了并行计算,要充分利用CPU的Core计算资源

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

RDD分区的原则是使得分区的个数尽量等于集群中的CPU核心(core)数目，这样可以充分利用CPU的计算资源

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

2.4 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操作

2.5 filter

对RDD中的每一个元素进行过滤,返回true留下,false过滤掉

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

//过滤出大于等于10的元素

val rdd3 = rdd2.filter(_ >= 10)

rdd3.collect

2.6 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)

2.7 sortBy

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

val rdd2 = rdd1.sortBy(x=>x,true) //默认升序

rdd2.collect

val rdd2 = rdd1.sortBy(x=>x+"",true)//变成了字符串,结果为字典顺序

rdd2.collect

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

注意类型要一致

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

val rdd2 = sc.parallelize(List(1, 2, 3, 4))

//求并集不会去重

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))

2.9 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))

2.10 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)))

2.11 cogroup(了解)

先在RDD内部按照key分组,再在多个RDD间按照key分组

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[(String, (Iterable[Int], Iterable[Int]))] = Array((tom,(CompactBuffer(1, 2),CompactBuffer(1))), (jerry,(CompactBuffer(3),CompactBuffer(2))), (shuke,(CompactBuffer(),CompactBuffer(2))), (kitty,(CompactBuffer(2),CompactBuffer())))

2.12 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)))

2.13 reduce--注意是Action算子

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

//reduce聚合

val result = rdd1.reduce(_ + _) //第一_ 上次一个运算的结果,第二个_ 这一次进来的元素

2.14 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))

2.15 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

注意：

repartition可以增加和减少rdd中的分区数，

coalesce默认减少rdd分区数，增加rdd分区数不会生效。

不管增加还是减少分区数原rdd分区数不变

应用场景:

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

2.16 collect

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

rdd1.collect

2.17 count

求RDD中最外层元素的个数

rdd1.count

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

2.18 distinct

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

rdd.distinct.collect

2.19 top

//取出最大的前N个

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

rdd1.top(2)

2.20 take

//按照原来的顺序取前N个

rdd1.take(2)

//需求:取出最小的2个

rdd1.sortBy(x=>x,true).take(2)

2.21 first

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

rdd1.first

2.22 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)

2.23 mapValues

val rdd1 = sc.parallelize(List((1,10),(2,20),(3,30)))

val rdd2 = rdd1.mapValues(_*2).collect //_表示每一个value ,key不变,将函数作用于value

//Array[(Int, Int)] = Array((1,20), (2,40), (3,60))

2.24 collectAsMap

val rdd = sc.parallelize(List(("a", 1), ("b", 2)))

rdd.collectAsMap

//scala.collection.Map[String,Int] = Map(b -> 2, a -> 1)

2.25 面试题：foreach和foreachPartition

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

rdd1.foreach(x => println(x*100)) //把函数传给各个分区,在分区内循环遍历该分区中的元素 //x每个元素,即一个一个的数字

rdd1.foreachPartition(x => println(x.reduce(_ + _))) //把各个分区传递给函数执行 //x是每个分区

应用场景:

比如在函数中要将RDD中的元素保存到数据库

foreach:会将函数作用到RDD中的每一条数据,那么有多少条数据,操作数据库连接的开启关闭就得执行多少次

foreachPartition:将函数作用到每一个分区,那么每一个分区执行一次数据库连接的开启关闭,有几个分区就会执行数据库连接开启关闭

2.26 面试题：map和mapPartitions

将每一个分区传递给函数

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

rdd1.mapPartitions(x=>x.map(y=>y*2)).collect //x是每个分区,y是分区中的每个元素

2.27 扩展：mapPartitionsWithIndex

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

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

//该函数的功能是将对应分区中的数据取出来，并且带上分区编号

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:0, val: 4],

[partID:1, val: 5], [partID:1, val: 6], [partID:1, val: 7], [partID:1, val: 8], [partID:1, val: 9]

)

2.28 扩展：aggregate

聚合

//0表示初始值,每个分区会有初始值,汇总的时候也有初始值

//第一个函数:表示各个分区如何操作

//第二个函数:表示对分区如何汇总操作

val result1: Int = rdd1.aggregate(0)(_+_,_+_) //45

val result2: Int = rdd1.aggregate(10)(_+_,_+_) //75

println(result1)

println(result2)

//源码中的注释:

@param zerovalue每个分区的累积结果的初始值

*`seqop`运算符，以及合并结果的初始值

*"combop"运算符的不同分区-这通常是中性元素（例如，"nil"用于列表连接，或"0"用于求和）

*@param seqop用于在分区内累积结果的运算符

*@param combop用于组合来自不同分区的结果的关联运算符

2.29 扩展：aggregateByKey

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

def func2(index: Int, iter: Iterator[(String, Int)]) : Iterator[String] = {

iter.map(x => "[partID:" + index + ", val: " + x + "]")

}

pairRDD.mapPartitionsWithIndex(func2).collect

//Array[String] = 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))

pairRDD.aggregateByKey(100)(math.max(_, _), _ + _).collect

2.30 扩展：combineByKey

val rdd1 = sc.textFile("hdfs://node-01:9000/wordcount/input/words.txt").flatMap(_.split(" ")).map((_, 1))

//Array((hello,1), (me,1), (hello,1), (you,1), (hello,1), (her,1))

val rdd2 = rdd1.combineByKey(x => x, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)

//val rdd2 = rdd1.combineByKey(_, _+_, _+_) //错误! 能省则省,不能省就不要偷懒了

rdd2.collect

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

https://blog.csdn.net/high2011/article/details/78852182

小练习

●需求

给定一个键值对RDD

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

key表示图书名称，

value表示某天图书销量，

请计算每个键对应的平均值，也就是计算每种图书的每天平均销量。最终结果:("spark",4),("hadoop",5)

●答案

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))

写成一行

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

.groupByKey().map(t=>(t._1,t._2.sum / t._2.size)).collect

总结

●分类

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

●注意:

RDD不实际存储真正要计算的数据，而只是记录了RDD的转换关系(调用了什么方法，传入什么函数，依赖哪些RDD，分区器是什么，数量块来源机器列表)

RDD中的所有转换操作都是延迟执行(懒执行)的，也就是说并不会直接计算。

只有当发生Action操作的时候，这些转换才会真正运行。