spark常用的算子

迈不过友情╰ 2022-05-27 02:21 204阅读 0赞

http://homepage.cs.latrobe.edu.au/zhe/ZhenHeSparkRDDAPIExamples.html
    
    map是对每个元素操作, mapPartitions是对其中的每个partition操作
    
    -------------------------------------------------------------------------------------------
    -------------------------------------------------------------------------------------------
    mapPartitionsWithIndex : 把每个partition中的分区号和对应的值拿出来, 看源码
    val func = (index: Int, iter: Iterator[(Int)]) => {
      iter.toList.map(x => "[partID:" +  index + ", val: " + x + "]").iterator
    }
    val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2)
    rdd1.mapPartitionsWithIndex(func).collect
    
    -------------------------------------------------------------------------------------------
    -------------------------------------------------------------------------------------------
    
    
    def func1(index: Int, iter: Iterator[(Int)]) : Iterator[String] = {
      iter.toList.map(x => "[partID:" +  index + ", val: " + x + "]").iterator
    }
    val rdd1 = sc.parallelize(List(1,2,3,4,5,6,7,8,9), 2)
    rdd1.mapPartitionsWithIndex(func1).collect
    
    aggregate
    ###是action操作, 第一个参数是初始值, 二:是2个函数[每个函数都是2个参数(第一个参数:先对个个分区进行合并, 第二个:对个个分区合并后的结果再进行合并), 输出一个参数]
    ###0 + (0+1+2+3+4   +   0+5+6+7+8+9)
    rdd1.aggregate(0)(_+_, _+_)
    rdd1.aggregate(0)(math.max(_, _), _ + _)
    ###5和1比, 得5再和234比得5 --> 5和6789比,得9 --> 5 + (5+9)
    rdd1.aggregate(5)(math.max(_, _), _ + _)

rdd1.reduce(math.max(_,_))//也可以求最大的数
    
    val rdd2 = sc.parallelize(List("a","b","c","d","e","f"),2)
    def func2(index: Int, iter: Iterator[(String)]) : Iterator[String] = {
      iter.toList.map(x => "[partID:" +  index + ", val: " + x + "]").iterator
    }
    rdd2.aggregate("")(_ + _, _ + _)
    rdd2.aggregate("=")(_ + _, _ + _)
    
    //
    
    val rdd3 = sc.parallelize(List("12","23","345","4567"),2)
    rdd3.aggregate("")((x,y) => math.max(x.length, y.length).toString, (x,y) => x + y)
    
    val rdd4 = sc.parallelize(List("12","23","345",""),2)
    rdd4.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y)
    
    val rdd5 = sc.parallelize(List("12","23","","345"),2)
    rdd5.aggregate("")((x,y) => math.min(x.length, y.length).toString, (x,y) => x + y)
    
    -------------------------------------------------------------------------------------------
    -------------------------------------------------------------------------------------------
    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.toList.map(x => "[partID:" +  index + ", val: " + x + "]").iterator
    }
    pairRDD.mapPartitionsWithIndex(func2).collect
    pairRDD.aggregateByKey(0)(math.max(_, _), _ + _).collect
    pairRDD.aggregateByKey(100)(math.max(_, _), _ + _).collect
    
    -------------------------------------------------------------------------------------------
    -------------------------------------------------------------------------------------------
    checkpoint
    sc.setCheckpointDir("hdfs://node-1.itcast.cn:9000/ck")
    val rdd = sc.textFile("hdfs://node-1.itcast.cn:9000/wc").flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_)
    rdd.checkpoint
    rdd.isCheckpointed
    rdd.count
    rdd.isCheckpointed
    rdd.getCheckpointFile
    
    -------------------------------------------------------------------------------------------
    -------------------------------------------------------------------------------------------
    coalesce, repartition
    val rdd1 = sc.parallelize(1 to 10, 10)
    val rdd2 = rdd1.coalesce(2, false) //设置为false的时候可以分区
    rdd2.partitions.length
    
    -------------------------------------------------------------------------------------------
    -------------------------------------------------------------------------------------------
    
    val rdd = sc.parallelize(List(("a", 1), ("b", 2)))
    rdd.collectAsMap
    结果:collectAsMap : Map(b -> 2, a -> 1) 把list转换成map
    -------------------------------------------------------------------------------------------
    -------------------------------------------------------------------------------------------
    combineByKey : 和reduceByKey是相同的效果
    ###第一个参数x:原封不动取出来(分组之后的第一个值), 第二个参数:是函数, 局部运算, 第三个:是函数, 对局部运算后的结果再做运算
    ###每个分区中每个key中value中的第一个值, (hello,1)(hello,1)(good,1)-->(hello(1,1),good(1))-->x就相当于hello的第一个1, good中的1
    val rdd1 =  sc.textFile("hdfs://hadoop-01:9000/user/wc").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect
    val rdd2 =  sc.textFile("hdfs://hadoop-01:9000/user/wc").flatMap(_.split(" ")).map((_,1)).combineByKey(x=>x,(m:Int,n:Int)=>m+n,(a:Int,b:Int)=>a+b).collect
     val rdd = sc.textFile("hdfs://hadoop-01:9000/user/wc").flatMap(_.split(" ")).map((_,1)).combineByKey(x=>x+10,(m:Int,n:Int)=>m+n,(a:Int,b:Int)=>a+b).collect
    val rdd2 = rdd1.combineByKey(x => x, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)
    rdd1.collect
    rdd2.collect
    
    ###当input下有3个文件时(有3个block块, 不是有3个文件就有3个block, ), 每个会多加3个10
    val rdd3 = rdd1.combineByKey(x => x + 10, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)
    rdd3.collect
    
    
    
    
    //把rdd4的数据按照rdd5来进行分区
    val rdd4 = sc.parallelize(List("dog","cat","gnu","salmon","rabbit","turkey","wolf","bear","bee"), 3)
    val rdd5 = sc.parallelize(List(1,1,2,2,2,1,2,2,2), 3)
    val rdd6 = rdd5.zip(rdd4)
    rdd6的结果:rdd6: Array[(Int, String)] = Array((1,dog), (1,cat), (2,gnu), (2,salmon), (2,rabbit), (1,turkey), (2,wolf), (2,bear), (2,bee))
    
    val rdd7 = rdd6.combineByKey(List(_), (x: List[String], y: String) => x :+ y, (m: List[String], n: List[String]) => m ++ n).collect
    
    结果:rdd7:Array[(Int, List[String])] = Array((1,List(dog, cat, turkey)), (2,List(gnu, salmon, rabbit, wolf, bear, bee)))
    
    
    
    -------------------------------------------------------------------------------------------
    -------------------------------------------------------------------------------------------
    countByKey 
    
    val rdd1 = sc.parallelize(List(("a", 1), ("b", 2), ("b", 2), ("c", 2), ("c", 1)))
    rdd1.countByKey  //结果scala.collection.Map[String,Long] = Map(b -> 2, a -> 1, c -> 2)  把相同key出现的个数累计相加并转为Map
    
    rdd1.countByValue //是把想同的keyValue一起出现的个数相加并转为Map
    //结果 scala.collection.Map[(String, Int),Long] = Map((b,2) -> 2, (c,2) -> 1, (a,1) -> 1, (c,1) -> 1)
    
    
    -------------------------------------------------------------------------------------------
    -------------------------------------------------------------------------------------------
    filterByRange
    
    val rdd1 = sc.parallelize(List(("e", 5), ("c", 3), ("d", 4), ("c", 2), ("a", 1)))
    val rdd2 = rdd1.filterByRange("b", "d").collect  //截取b到d之间的数据 包头也包尾袄
    结果: Array[(String, Int)] = Array((c,3), (d,4), (c,2)) 
    
    -------------------------------------------------------------------------------------------
    -------------------------------------------------------------------------------------------
    
    val rdd3 = sc.parallelize(List(("a", "1 2"), ("b", "3 4")))
    val rdd4 = rdd3.flatMapValues(_.split(" ")).collect       //先把元组的value拿出来进行压平切分
    结果:Array[(String, String)] = Array((a,1), (a,2), (b,3), (b,4))   
    
    -------------------------------------------------------------------------------------------
    -------------------------------------------------------------------------------------------
    foldByKey //跟aggregate 比较 aggregate先是局部操作 再做全局操作 效率高 两个参数 rdd1.aggregate(0)(_+_,_+_)
    
    val rdd1 = sc.parallelize(List("dog", "wolf", "cat", "bear"), 2)
    val rdd2 = rdd1.map(x => (x.length, x)) //结果  Array[(Int, String)] = Array((3,dog), (4,wolf), (3,cat), (4,bear))  
    val rdd3 = rdd2.foldByKey(" ")(_+_) //做的是根据Key进行分组归类的操作
     //结果 Array[(Int, String)] = Array((4," wolf bear"), (3," dog cat"))     
    
    val rdd = sc.textFile("hdfs://hadoop-01:9000/user/wc").flatMap(_.split(" ")).map((_, 1))
    rdd.foldByKey(0)(_+_)
    
    -------------------------------------------------------------------------------------------
    -------------------------------------------------------------------------------------------
    foreachPartition
    val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9), 3)
    rdd1.foreachPartition(x => println(x.reduce(_ + _)))
    
    -------------------------------------------------------------------------------------------
    -------------------------------------------------------------------------------------------
    keyBy : 以传入的参数做key
    val rdd1 = sc.parallelize(List("dog", "salmon", "salmon", "rat", "elephant"), 3)
    val rdd2 = rdd1.keyBy(_.length)    //获得key的长度
    结果: Array[(Int, String)] = Array((3,dog), (6,salmon), (6,salmon), (3,rat), (8,elephant))
    val rdd3 = rdd1.keyBy(_(0)).collect  //根据传入的参数作为下标数来决定key  0 就是第一位  如果1的话dog 的key就为e
    结果: Array[(Char, String)] = Array((d,dog), (s,salmon), (s,salmon), (r,rat), (e,elephant))
    rdd2.collect
    
    -------------------------------------------------------------------------------------------
    -------------------------------------------------------------------------------------------
    keys values
    val rdd1 = sc.parallelize(List("dog", "tiger", "lion", "cat", "panther", "eagle"), 2)
    val rdd2 = rdd1.map(x => (x.length, x))
    rdd2.keys.collect  结果 Array[Int] = Array(3, 5, 4, 3, 7, 5)  
    rdd2.values.collect 结果  Array[String] = Array(dog, tiger, lion, cat, panther, eagle)
    
    
    -------------------------------------------------------------------------------------------
    -------------------------------------------------------------------------------------------
    mapPartitions