RDD常用的算子操作, map和mapPartitions区别, groupByKey和groupBy区别, repartition和coalesce区别, 广播大变量broadcast 09

痛定思痛。 2021-12-17 13:09 301阅读 0赞

# 1. filter #

filter 是对RDD中的每个元素都执行一个指定的函数来过滤产生一个新的RDD。 任何原RDD中的元素在新RDD中都有且只有一个元素与之对应。

val rdd = sc.parallelize(List(1,2,3,4,5,6))  
    val filterRdd = rdd.filter(_ > 5)
    filterRdd.collect() //返回所有大于5的数据的一个Array， Array(6,8,10,12)

# 2. flatMap #

与map类似，区别是原RDD中的元素经map处理后只能生成一个元素，而原RDD中的元素经flatmap处理后可生成多个元素来构建新RDD。 举例：对原RDD中的每个元素x产生y个元素（从1到y，y为元素x的值）

scala> val a = sc.parallelize(1 to 4, 2)
    scala> val b = a.flatMap(x => 1 to x)
    scala> b.collect
    res12: Array[Int] = Array(1, 1, 2, 1, 2, 3, 1, 2, 3, 4)

# 3. map和mapParitions区别 #

## 3.1 map ##

*  map是对RDD中的每个元素都执行一个指定的函数来产生一个新的RDD.
 *  任何原RDD中的元素在新的RDD中都有且只有一个元素与之对应.

scala> val a = sc.parallelize(1 to 9, 3)
    scala> val b = a.map(x => x*2)
    scala> a.collect
    res10: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)
    scala> b.collect
    res11: Array[Int] = Array(2, 4, 6, 8, 10, 12, 14, 16, 18)

**上述例子中把原RDD中每个元素都乘以2来产生一个新的RDD。**

## 3.2 mapParitions ##

*  mapParitions是map的一个变种.map的输入函数是应用于RDD中每个元素.
 *  而mapParitions的输入函数是应用于每个分区,也就是把每个分区中的内容作为整体来处理的.
 *

scala> val a = sc.parallelize(1 to 9, 3)
    scala> def myfunc[T](iter: Iterator[T]) : Iterator[(T, T)] = { 
      var res = List[(T, T)]() 
      var pre = iter.next 
    while (iter.hasNext) { 
        val cur = iter.next
        res.::=(pre, cur)
          pre = cur  } 
      res.iterator
    }

*  上述例子中的函数myfunc是把分区中一个元素和它的下一个元素组成一个Tuple。因为分区中最后一个元素没有下一个元素了，所以(3,4)和(6,7)不在结果中。
 *  mapPartitions还有些变种，比如mapPartitionsWithContext，它能把处理过程中的一些状态信息传递给用户指定的输入函数。还有mapPartitionsWithIndex，它能把分区的index传递给用户指定的输入函数。

# 4. 交集intersection, 并集union #

*  都依赖于两个rdd.

val rdd1 = sc.parallelize(List(5,6,4,3))
    val rdd2 = sc.parallelize(List(1,2,3,4))
    //求并集
    val rdd3 = rdd1.union(rdd2)
    //求交集
    val rdd4 = rdd1.intersection(rdd2)
    //去重
    rdd3.distinct.collect
    rdd4.collect

# 5. join() 左连接 #

*  join() **rdd两两左连接**
 *  rdd1.join(rdd2) --> Array(k, (v,w))
 *  以rdd1为基准

val rdd7: RDD[(String, Any)] = sc.parallelize(List(("tom", "aa"),("tom", 1), ("tom", "bb"),("jerry", 3), ("kitty", 2)))
    val rdd8 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("shuke", 2)))
    //join
     val join: RDD[(String, (Any, Int))] = rdd7.join(rdd8)
     val c7: Array[(String, (Any, Int))] = join.collect()
    //结果 ArrayBuffer((tom,(aa,1)), (tom,(1,1)), (tom,(bb,1)), (jerry,(3,2)))

# 6. cogroup和groupByKey区别 #

## 6.1 cogroup() ##

*  同一个rdd中, **相同k的value迭代组成迭代器k,(Iterator\[v\],Iterator\[w\])**
 *  **依赖于两个rdd**
 *  举例: **rdd.cogroup(rdd2)**

val rdd9: RDD[(String, Any)] = sc.parallelize(List(("tom", "aa"),("tom", 1), ("tom", "bb"),("jerry", 3), ("kitty", 2)))
     val rdd10 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("shuke", 2)))
     val cogroup: RDD[(String, (Iterable[Any], Iterable[Int]))] = rdd9.cogroup(rdd10)
        val c9: Array[(String, (Iterable[Any], Iterable[Int]))] = cogroup.collect()
    //结果
    ArrayBuffer((tom,(CompactBuffer(aa, 1, bb),CompactBuffer(1))), (jerry,(CompactBuffer(3),CompactBuffer(2))), (shuke,(CompactBuffer(),CompactBuffer(2))), (kitty,(CompactBuffer(2),CompactBuffer())))

## 6.2 groupByKey() ##

*  **不依赖多个RDD,一个rdd执行起来的算子**
 *  求一个集合的相同k合并后的value, shuffle阶段, 得到的value是迭代器 rdd.groupByKey()得到的是 Array(k, Iterator\[v\])

val rdd7: RDD[(String, Any)] = sc.parallelize(List(("tom", "aa"),("tom", 1), ("tom", "bb"),("jerry", 3), ("kitty", 2))) 
    val key: RDD[(String, Iterable[Any])] = rdd7.groupByKey()
     val c77: Array[(String, Iterable[Any])] = key.collect()
    //结果 
    ArrayBuffer((tom,CompactBuffer(aa, 1, bb)), (jerry,CompactBuffer(3)), (kitty,CompactBuffer(2)))

# 7. groupBy() #

*  \*\*groupBy()\*\*更加灵活 返回的是 k, Iterator(k,v)

ArrayBuffer((tom,CompactBuffer((tom,aa), (tom,1), (tom,bb))), (jerry,CompactBuffer((jerry,3))), (kitty,CompactBuffer((kitty,2))))

# 8. reduce #

val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5))
    //reduce聚合
    val rdd2 = rdd1.reduce(_ + _)
    rdd2.collect

# 9. reduceByKey, sortByKey #

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)
    //按key进行聚合
    val rdd4 = rdd3.reduceByKey(_ + _)
    rdd4.collect
    //按value的降序排序
    val rdd5 = rdd4.map(t => (t._2, t._1)).sortByKey(false).map(t => (t._2, t._1))
    rdd5.collect

# 10. 重点, 分区划分->repartition和coalesce区别 #

*  **reparition可以增加和减少RDD中的分区数**
 *  **coalesce只能减少RDD分区数,增加RDD分区数不会生效.**

val rdd1 = sc.parallelize(1 to 10,3)
    //利用repartition改变rdd1分区数
    //减少分区
    rdd1.repartition(2).partitions.size
    //增加分区
    rdd1.repartition(4).partitions.size
    //利用coalesce改变rdd1分区数
    //减少分区
    rdd1.coalesce(2).partitions.size

# 11. Spark中广播变量 #

*  一个executor有多个task,每个task都会被加载数据,为了减轻内部压力,将公共的数据广播出去
 *  广播的是**字典数据(也就是不会频繁改变的数据)**
 *  放到每一个executor中一个公共的位置
 *  广播的数据是**Action动作后的数据**

//广播数据(广播action动作后的数据)
    val broadcast: Broadcast[Array[(String, String, String, String)]] = sc.broadcast(collect)
    //取数据
    val valueArr: Array[(String, String, String, String)] = broadcast.value