RDD-API 创建RDD三种方法 RDD的方法/算子分类 Transformation转换算子 Action动作算子 统计操作 基础练习[快速演示] 准备工作 案例 1. WordCount 2. 创建RDD 3. 查看该RDD的分区数量 4. map 5. filter 6. flatmap 7. sortBy 8. 交集、并集、差集、笛卡尔积 9. Join 10. groupbykey 11. cogroup[了解] 12.groupBy 13.reduce 14. reducebykey 15. repartition 16. collect 17.count 18.distinct 19. top 20. take 21. first 22. keys、values 23. mapValues 24. collectAsMap 25. 扩展：mapPartitionsWithIndex（同时获取分区号） 26. 扩展：aggregate 27.扩展：combineByKey 28. 扩展：aggregateByKey 小练习 总结 ★★★★★--------面试题--------★★★★★ reduceByKey 与reduce 分别是Transformation还是Action 面试题：foreach和foreachPartition 面试题：map和mapPartitions 面试题小总结 创建RDD三种方法 由外部存储系统的数据集创建，包括本地的文件系统，还有所有Hadoop支持的数据集，比如HDFS、Cassandra、HBase等 通过已有的RDD经过算子转换生成新的RDD 由一个已经存在的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方法
SparkCore:RDD-API史上最详细操作(内含面试题)创建RDD三种方法RDD的方法/算子分类总结★★★★★--------面试题--------★★★★★面试题小总结 RDD的方法/算子分类 分类
RDD的算子分为两类:
Transformation转换操作:返回一个新的RDD Action动作操作:返回值不是RDD(无返回值或返回其他的) SparkCore:RDD-API史上最详细操作(内含面试题)创建RDD三种方法RDD的方法/算子分类总结★★★★★--------面试题--------★★★★★面试题小总结 注意:
RDD不实际存储真正要计算的数据，而是记录了数据的位置在哪里，数据的转换关系(调用了什么方法，传入什么函数)
RDD中的所有转换都是惰性求值/延迟执行的，也就是说并不会直接计算。只有当发生一个要求返回结果给Driver的Action动作时，这些转换才会真正运行。
之所以使用惰性求值/延迟执行，是因为这样可以在Action时对RDD操作形成DAG有向无环图进行Stage的划分和并行优化，这种设计让Spark更加有效率地运行 Transformation转换算子 转换 含义 map(func) 返回一个新的RDD，该RDD由每一个输入元素经过func函数转换后组成 filter(func) 返回一个新的RDD，该RDD由经过func函数计算后返回值为true的输入元素组成 flatMap(func) 类似于map，但是每一个输入元素可以被映射为0或多个输出元素(所以func应该返回一个序列，而不是单一元素) mapPartitions(func) 类似于map，但独立地在RDD的每一个分片上运行，因此在类型为T的RDD上运行时，func的函数类型必须是Iterator[T] => Iterator[U] mapPartitionsWithIndex(func) 类似于mapPartitions，但func带有一个整数参数表示分片的索引值，因此在类型为T的RDD上运行时，func的函数类型必须是(Int, Interator[T]) => Iterator[U] sample(withReplacement, fraction, seed) 根据fraction指定的比例对数据进行采样，可以选择是否使用随机数进行替换，seed用于指定随机数生成器种子 union(otherDataset) 对源RDD和参数RDD求并集后返回一个新的RDD intersection(otherDataset) 对源RDD和参数RDD求交集后返回一个新的RDD distinct([numTasks])) 对源RDD进行去重后返回一个新的RDD groupByKey([numTasks]) 在一个(K,V)的RDD上调用，返回一个(K, Iterator[V])的RDD reduceByKey(func, [numTasks]) 在一个(K,V)的RDD上调用，返回一个(K,V)的RDD，使用指定的reduce函数，将相同key的值聚合到一起，与groupByKey类似，reduce任务的个数可以通过第二个可选的参数来设置 aggregateByKey(zeroValue)(seqOp, combOp, [numTasks]) sortByKey([ascending], [numTasks]) 在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD sortBy(func,[ascending], [numTasks]) 与sortByKey类似，但是更灵活 join(otherDataset, [numTasks]) 在类型为(K,V)和(K,W)的RDD上调用，返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD cogroup(otherDataset, [numTasks]) 在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable,Iterable))类型的RDD cartesian(otherDataset) 笛卡尔积 pipe(command, [envVars]) 对rdd进行管道操作 coalesce(numPartitions) 减少 RDD 的分区数到指定值。在过滤大量数据之后，可以执行此操作 repartition(numPartitions) 重新给 RDD 分区 Action动作算子 动作 含义 reduce(func) 通过func函数聚集RDD中的所有元素，这个功能必须是可交换且可并联的 collect() 在驱动程序中，以数组的形式返回数据集的所有元素 count() 返回RDD的元素个数 first() 返回RDD的第一个元素(类似于take(1)) take(n) 返回一个由数据集的前n个元素组成的数组 takeSample(withReplacement,num, [seed]) 返回一个数组，该数组由从数据集中随机采样的num个元素组成，可以选择是否用随机数替换不足的部分，seed用于指定随机数生成器种子 takeOrdered(n, [ordering]) 返回自然顺序或者自定义顺序的前 n 个元素 saveAsTextFile(path) 将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统，对于每个元素，Spark将会调用toString方法，将它装换为文件中的文本 saveAsSequenceFile(path) 将数据集中的元素以Hadoop sequencefile的格式保存到指定的目录下，可以使HDFS或者其他Hadoop支持的文件系统。 saveAsObjectFile(path) 将数据集的元素，以 Java 序列化的方式保存到指定的目录下 countByKey() 针对(K,V)类型的RDD，返回一个(K,Int)的map，表示每一个key对应的元素个数。 foreach(func) 在数据集的每一个元素上，运行函数func进行更新。 foreachPartition(func) 在数据集的每一个分区上，运行函数func 统计操作 算子 含义 count 个数 mean 均值 sum 求和 max 最大值 min 最小值 variance 方差 sampleVariance 从采样中计算方差 stdev 标准差:衡量数据的离散程度 sampleStdev 采样的标准差 stats 查看统计结果 基础练习[快速演示] 准备工作 集群模式启动 启动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
或本地模式启动 /export/servers/spark/bin/spark-shell 案例 1. WordCount val res = sc.textFile("hdfs://node01:8020/wordcount/input/words.txt")
.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
//上面的代码不会立即执行,因为都是Transformation转换操作
//下面的代码才会真正的提交并执行,因为是Action动作/行动操作
res.collect
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))
3. 查看该RDD的分区数量 //没有指定分区数,默认值是2
sc.parallelize(List(5,6,4,7,3,8,2,9,1,10)).partitions.length
//指定了分区数为3
sc.parallelize(List(5,6,4,7,3,8,2,9,1,10),3).partitions.length
//默认为2
sc.textFile("hdfs://node01:8020/wordcount/input/words.txt").partitions.length
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
4. map 对RDD中的每一个元素进行操作并返回操作的结果
//通过并行化生成rdd
val rdd1 = sc.parallelize(List(5, 6, 4, 7, 3, 8, 2, 9, 1, 10))
//对rdd1里的每一个元素
//_.* (x=>x*2) x是List里所有的数据
rdd1.map(_ * 2).collect //collect方法表示收集,是action操作
5. filter 注意:函数中返回True的被留下,返回False的被过滤掉
val rdd2 = sc.parallelize(List(5, 6, 4, 7, 3, 8, 2, 9, 1, 10))
//val rdd3 = rdd2.filter(x=> x >= 10)
val rdd3 = rdd2.filter(_ >= 10)
rdd3.collect //10
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)
7. 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...
8. 交集、并集、差集、笛卡尔积 注意类型要一致
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))
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)))
图解 SparkCore:RDD-API史上最详细操作(内含面试题)创建RDD三种方法RDD的方法/算子分类总结★★★★★--------面试题--------★★★★★面试题小总结 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)))
图解 SparkCore:RDD-API史上最详细操作(内含面试题)创建RDD三种方法RDD的方法/算子分类总结★★★★★--------面试题--------★★★★★面试题小总结 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))
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)))
11. 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())))
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)))
13.reduce val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5))
//reduce聚合((a,b)+>a+b)
val result = rdd1.reduce(_ + _)
// 第一_ 上次一个运算的结果,第二个_ 这一次进来的元素
14. 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))
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
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”)
16. collect val rdd1 = sc.parallelize(List(6,1,2,3,4,5), 2)
rdd1.collect
17.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
18.distinct val rdd = sc.parallelize(Array(1,2,3,4,5,5,6,7,8,1,2,3,4), 3)
rdd.distinct.collect
19. top //取出最大的前N个
val rdd1 = sc.parallelize(List(3,6,1,2,4,5))
rdd1.top(2)
20. take //按照原来的顺序取前N个
rdd1.take(2) //3 6
//需求:取出最小的2个
rdd1.sortBy(x=>x,true).take(2)
21. first //按照原来的顺序取前第一个
rdd1.first
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)
23. mapValues mapValues表示对RDD中的元素进行操作,Key不变,Value变为操作之后
val rdd1 = sc.parallelize(List((1,10),(2,20),(3,30)))
val rdd2 = rdd1.mapValues(_*2).collect //_表示每一个value ,key不变,将函数作用于value
//(1,20),(2,40),(3,60)
24. collectAsMap 转换成Map
val rdd = sc.parallelize(List(("a", 1), ("b", 2)))
rdd.collectAsMap
//scala.collection.Map[String,Int] = Map(b -> 2, a -> 1)
25. 扩展：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]
)
26. 扩展：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
27.扩展：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
28. 扩展：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动作操作
●注意:
RDD不实际存储真正要计算的数据，而只是记录了RDD的转换关系(调用了什么方法，传入什么函数，依赖哪些RDD，分区器是什么，数量块来源机器列表)
RDD中的所有转换操作都是延迟执行(懒执行)的，也就是说并不会直接计算。只有当发生Action操作的时候，这些转换才会真正运行。
★★★★★--------面试题--------★★★★★ reduceByKey 与reduce 分别是Transformation还是Action reduceByKey是Transformation还是Action? --Transformation
reduce是Transformation还是Action? --Action
面试题：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是每个分区
注意: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)
rdd1.mapPartitions(x=>x.map(y=>y*2)).collect //x是每一个分区,y是分区中的元素
面试题小总结 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作用于每个分区