哪个网站做外贸假发好,wordpress源码站整站源码,aso关键词覆盖优化,做影视网站需要多少钱掌握Spark高级算子在代码中的使用 相同点分析 三个函数的共同点#xff0c;都是Transformation算子。惰性的算子。 不同点分析 map函数是一条数据一条数据的处理#xff0c;也就是#xff0c;map的输入参数中要包含一条数据以及其他你需要传的参数。 mapPartitions函数是一个…掌握Spark高级算子在代码中的使用 相同点分析 三个函数的共同点都是Transformation算子。惰性的算子。 不同点分析 map函数是一条数据一条数据的处理也就是map的输入参数中要包含一条数据以及其他你需要传的参数。 mapPartitions函数是一个partition数据一起处理也即是说mapPartitions函数的输入是一个partition的所有数据构成的“迭代器”然后函数里面可以一条一条的处理在把所有结果按迭代器输出。也可以结合yield使用效果更优。 rdd的mapPartitions是map的一个变种它们都可进行分区的并行处理。 两者的主要区别是调用的粒度不一样map的输入变换函数是应用于RDD中每个元素而mapPartitions的输入函数是应用于每个分区。 mapPartitionsWithIndex函数其实和mapPartitions函数区别不大因为mapPartitions背后调的就是mapPartitionsWithIndex函数只是一个参数被close了。mapPartitionsWithIndex的函数可以或得partition索引号 假设一个rdd有10个元素分成3个分区。如果使用map方法map中的输入函数会被调用10次而使用mapPartitions方法的话其输入函数会只会被调用3次每个分区调用1次。 mapPartitionsWithIndex则是带上分区下标进行操作。 1、mapPartitionWithIndex import org.apache.spark.{SparkConf,SparkContext}
object mapPartitionWithIndex {def getPartInfo:(Int,Iterator[Int]) Iterator[String] (index:Int,iter:Iterator[Int]) {
iter.map(x [ PartId index , elems: x ])
}def main(args: Array[String]): Unit {
val conf new SparkConf().setMaster(local).setAppName(RddMapPartitionsWithIndexDemo)
val sc new SparkContext(conf)val rdd1 sc.parallelize(List(1,2,3,4,5,9,6,7,8),numSlices 3 )
val rdd2 rdd1.mapPartitionsWithIndex(getPartInfo)
rdd2.collect().foreach(println)}} 2、aggregate 首先我们来创建一个 RDD scala val rdd1 sc.parallelize(1 to 5)
rdd1: org.apache.spark.rdd.RDD[Int] ParallelCollectionRDD[4] at parallelize at console:24scala rdd1.collect
warning: there was one feature warning; re-run with -feature for details
res24: Array[Int] Array(1, 2, 3, 4, 5) 这个 RDD 仅有 1 个分片包含 5 个数据 1, 2, 3, 4, 5 。 然后我们来应用一下 aggregate 方法。 在使用 aggregate 之前我们还是先定义两个要给 aggregate 当作输入参数的函数吧。 scala
// Entering paste mode (ctrl-D to finish)
def pfun1(p1: Int, p2: Int): Int {
p1 * p2
}// Exiting paste mode, now interpreting.
pfun1: (p1: Int, p2: Int)Intscalascala
// Entering paste mode (ctrl-D to finish)
def pfun2(p3: Int, p4: Int): Int {
p3 p4
}// Exiting paste mode, now interpreting.
pfun2: (p3: Int, p4: Int)Intscala 接着是第 2 个函数。就不再解释什么了。 然后终于可以开始应用我们的 aggregate 方法了。 scala rdd1.aggregate(3)(pfun1, pfun2)
res25: Int 363scala 输出结果是 363 这个结果是怎么算出来的呢 首先我们的 zeroValue 即初值是 3 。然后通过上面小节的介绍我们知道首先会应用 pfun1 函数因为我们这个 RDD 只有 1 个分片所以整个运算过程只会有一次 pfun1 函数调用。它的计算过程如下 首先用初值 3 作为 pfun1 的参数 p1 然后再用 RDD 中的第 1 个值即 1 作为 pfun1 的参数 p2 。由此我们可以得到第一个计算值为 3 * 1 3 。接着这个结果 3 被当成 p1 参数传入RDD 中的第 2 个值即 2 被当成 p2 传入由此得到第二个计算结果为 3 * 2 6 。以此类推整个 pfun1 函数执行完成以后得到的结果是 3 * 1 * 2 * 3 * 4 * 5 360 。这个 pfun1 的应用过程有点像是 “在 RDD 中滑动计算” 。 在 aggregate 方法的第 1 个参数函数 pfun1 执行完毕以后我们得到了结果值 360 。于是这个时候就要开始执行第 2 个参数函数 pfun2 了。 pfun2 的执行过程与 pfun1 是差不多的同样会将 zeroValue 作为第一次运算的参数传入在这里即是将 zeroValue 即 3 当成 p3 参数传入然后是将 pfun1 的结果 360 当成 p4 参数传入由此得到计算结果为 363 。因为 pfun1 仅有一个结果值所以整个 aggregate 过程就计算完毕了最终的结果值就是 363 。 import org.apache.spark.{SparkConf, SparkContext}object RddAggregateDemo {def main(args: Array[String]): Unit {
val conf new SparkConf().setMaster(local).setAppName(RddDemos)
val sc new SparkContext(conf)
val rdd1 sc.parallelize(List(12,23,345,),numSlices 2)
//下面代码等价与rdd1.aggregate()(xy) math.min(x.length,y.length)),(x,y)xy))
val result rdd1.aggregate()(func1,func2)
println(result)
}
//(x,y)math.min(x.Length,y.length)
def func1:(String,String) String (x:String,y:String) {
println(x: x ,x.len: x.length ,y: y ,y.len: y.length )
val ret math.min(x.length,y.length).toString
println(func1 ret: ret)
ret
}
//(xy) xy
def func2:(String,String)String(x:String,y:String) {
println( (xy))
xy
}
} 3、aggregateByKey 通过scala集合以并行化方式创建一个RDD scala val pairRdd sc.parallelize(List((“cat”,2),(“cat”,5),(“mouse”,4),(“cat”,12),(“dog”,12),(“mouse”,2)),2) pairRdd 这个RDD有两个区一个区中存放的是 (“cat”,2),(“cat”,5),(“mouse”,4) 另一个分区中存放的是 (“cat”,12),(“dog”,12),(“mouse”,2) 然后执行下面的语句 scala pairRdd.aggregateByKey(100)(math.max(_ , _), _ _ ).collect 结果 res0: Array[(String,Int)] Array((dog,100),(cat,200),(mouse,200) 下面是以上语句执行的原理详解 aggregateByKey的意思是按照key进行聚合 第一步将每个分区内key相同数据放到一起 分区一 (“cat”,(2,5)),(“mouse”,4) 分区二 (“cat”,12),(“dog”,12),(“mouse”,2) 第二步局部求最大值 对每个分区应用传入的第一个函数math.max(_ , _)这个函数的功能是求每个分区中每个key的最大值 这个时候要特别注意aggregateByKe(100)(math.max(_ , _),__)里面的那个100其实是个初始值 在分区一中求最大值的时候,100会被加到每个key的值中这个时候每个分区就会变成下面的样子 分区一 (“cat”,(2,5100)),(“mouse”,(4100)) 然后求最大值后变成 (“cat”,100), (“mouse”,100) 分区二 (“cat”,(12,100)),(“dog”,(12.100)),(“mouse”,(2,100)) 求最大值后变成 (“cat”,100),(“dog”,100),(“mouse”,100) 第三步整体聚合 将上一步的结果进一步的合成这个时候100不会再参与进来 最后结果就是 (dog,100),(cat,200),(mouse,200) 对PairRDD中相同的Key值进行聚合操作在聚合过程中同样使用了一个中立的初始值。和aggregate函数类似aggregateByKey返回值的类型不需要和RDD中value的类型一致。因为aggregateByKey是对相同Key中的值进行聚合操作所以aggregateByKey函数最终返回的类型还是PairRDD对应的结果是Key和聚合后的值而aggregate函数直接返回的是非RDD的结果。 import org.apache.spark.{SparkConf, SparkContext}object RddAggregateByKeyDemo {
def main(args: Array[String]): Unit {
val conf new SparkConf().setMaster(local).setAppName(RddAggregateByKeyDemo)
val sc new SparkContext(conf)
val pairRDD sc.parallelize(List((cat,2),(cat,5),(mouse,4),(cat,12),(dog,12),(mouse,2)),numSlices 2)
val rdd pairRDD.aggregateByKey(100)(func2,func2)
val resultArray rdd.collect
resultArray.foreach(println)
sc.stop()}
def func2(index:Int,iter:Iterator[(String,Int)]):Iterator[String] {
iter.map(x[partID: index ,val: x ])
}//(x,y)math.max(xy)
def func1:(Int,Int) Int (x:Int,y:Int) {
println(x: x , ,y: y )
val ret math.max(x,y)
println(func1 max: ret)
ret
}
//(xy) xy
def func2:(Int,Int)Int(x:Int,y:Int) {
println(func2 x : x ,y:y)
println(func2 ret (xy))
xy
}}
