makeRDD方法底层调用了parallelize方法 之所以使用惰性求值/延迟执行，是因为这样可以在Action时对RDD操作形成DAG有向无环图进行Stage的划分和并行优化，这种设计让Spark更加有效率地运行 启动Spark集群 启动spark-shell /export/servers/spark/bin/spark-shell //上面的代码不会立即执行,因为都是Transformation转换操作 对RDD中的每一个元素进行操作并返回操作的结果 注意:函数中返回True的被留下,返回False的被过滤掉 对RDD中的每一个元素进行先map再压扁,最后返回操作的结果 注意类型要一致 join(内连接)聚合具有相同key组成的value元组 groupByKey()的功能是，对具有相同键的值进行分组。 cogroup是先RDD内部分组,在RDD之间分组 根据指定的函数中的规则/key进行分组 注意reducebykey是转换算子 ★注意： ★应用场景: 功能：取分区中对应的数据时，还可以将分区的编号取出来，这样就可以知道数据是属于哪个分区的 ● 需求 给定一个键值对RDD ● 答案 ● 分类 ● 如何区分Transformation和Action ●注意: 应用场景:
RDD-API
创建RDD三种方法
val rdd1 = sc.textFile("hdfs://node01:8020/wordcount/input/words.txt")
val rdd2=rdd1.flatMap(_.split(" "))
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))
RDD的方法/算子分类
RDD的算子分为两类:
RDD不实际存储真正要计算的数据，而是记录了数据的位置在哪里，数据的转换关系(调用了什么方法，传入什么函数)
RDD中的所有转换都是惰性求值/延迟执行的，也就是说并不会直接计算。只有当发生一个要求返回结果给Driver的Action动作时，这些转换才会真正运行。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
查看统计结果
基础练习[快速演示]
准备工作
/export/servers/spark/sbin/start-all.sh
/export/servers/spark/bin/spark-shell
–master spark://node01:7077
–executor-memory 1g
–total-executor-cores 2
案例
1. WordCount
val res = sc.textFile("hdfs://node01:8020/wordcount/input/words.txt") .flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
//下面的代码才会真正的提交并执行,因为是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分区的原则是使得分区的个数尽量等于集群中的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 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
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
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
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)) 10. 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[了解]
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
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(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 小练习
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动作操作
返回值是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作用于每个分区
本网页所有视频内容由 imoviebox边看边下-网页视频下载, iurlBox网页地址收藏管理器 下载并得到。
ImovieBox网页视频下载器 下载地址: ImovieBox网页视频下载器-最新版本下载
本文章由: imapbox邮箱云存储,邮箱网盘,ImageBox 图片批量下载器,网页图片批量下载专家,网页图片批量下载器,获取到文章图片,imoviebox网页视频批量下载器,下载视频内容,为您提供.
阅读和此文章类似的: 全球云计算
官方软件产品操作指南 (170)
