0


Spark RDD的转换

按颜色区分转换:

  • 绿色是单 RDD 窄依赖转换
  • 黑色是多 RDD 窄依赖转换
  • 紫色是 KV 洗牌型转换
  • 黄色是重分区转换
  • 蓝色是特例的转换

单 RDD 窄依赖转换

MapPartitionRDD

这个 RDD 在第一次分析中已经分析过。简单复述一下:

  • 依赖列表:一个窄依赖,依赖上游 RDD
  • 分区列表:上游 RDD 的分区列表
  • 计算流程:映射关系(输入一个分区,返回一个迭代器)
  • 分区器 :上游 RDD 的分区器
  • 存储位置:上游 RDD 的优先位置 可见除了计算流程,其他都是上游 RDD 的内容。
  1. map 传入一个带“值到值”转化函数的迭代器(例如字符串到字符串长度)
  2. mapPartitions 传入一个“迭代器到迭代器”的转化函数,如果需要按分区做一些比较重的过程(例如数据库连接等)
  3. flatMap 传入一个“迭代器到迭代器的迭代器”的转化函数(例如,统计字母,“字符串的迭代器”到“‘字符的迭代器’的迭代器”)
  4. filter 传入了一个带“值到布尔值”筛选函数的迭代器

PartitionwiseSampledRDD

在分区中采样的RDD

  • 分区列表:在上游的分区的基础上包装一个采样过程,形成一个新的分区PartitionwiseSampledRDDPartition
  • 计算流程:采样器返回的迭代器
  • 其他成分:与上游 RDD 相同 PartitionwiseSampledRDD,有放回的采样用泊松采样器,无放回的采样用伯努利采样器,传给分区器。

多 RDD 窄依赖

UnionRDD

  • 依赖列表:每个上游 RDD 一个RangeDependency,每个RangeDependency依赖上游 RDD 的所有分区
  • 分区列表:每个上游 RDD 一个UnionPartition,构成列表
  • 计算流程:获得目标分区的迭代器
  • 分区器 :None
  • 存储位置:每个上游 RDD 的优先位置

CartesianRDD

笛卡尔积,是两个 RDD 每个数据都进行一次关联。下文中两个 RDD 的关联中,两个 RDD 分别称为 rdd1、rdd2。

  • 依赖列表:两个窄依赖组成的数组,分别依赖 rdd1、rdd2
  • 分区列表:“rdd1的分区数 乘以 rdd2的分区数”个分区
  • 计算流程:rdd1的一条记录与 rdd2的一条记录合成元组
  • 分区器 :None
  • 存储位置:rdd1、rdd2的存储位置的积

洗牌型转换

洗牌型转换,是多个 RDD 关联的的转换。

CoGroupedRDD

多个源 RDD 依据 key 关联,key 相同的合并,形成最终的目标 RDD。

  • 依赖列表:每个源 RDD 一个依赖,构成列表。如果源 RDD 的分区器与目标的分区器相同,则是1-to-1依赖,如果不同,则是洗牌依赖
  • 分区列表:目标 RDD 分区器指定的分区数量个CoGroupPartition,每个分区记录了数据来源分区。其中如果是洗牌依赖的数据源,需要洗牌过程,具体洗牌过程以后再分析
  • 计算流程:返回一个迭代器,迭代对象是 key 和 key 对应源分区迭代器的数组 组成的元祖
  • 分区器 :目标 RDD 的分区器
  • 存储位置:None

ShuffledRDD

同样是多个源 RDD 依据 key 关联,key 相同的做排序或聚合运算,形成最终的目标 RDD。

  • 依赖列表:一个洗牌依赖,依赖所有上游 RDD
  • 分区列表:目标 RDD 分区器指定的分区数量个ShuffledRDDPartition,每个分区只有一个编号(因为每个上游分区)
  • 计算流程:洗牌过程,具体洗牌过程以后再分析
  • 分区器 :目标 RDD 的分区器
  • 存储位置:None

除了这五个成员以外,还有另外几个重要的成员:序列化器、key 排序器、聚合器、map 端合并器,他们都将用于洗牌

其他

  • coalesce,是减少分区数量,可以在过滤之后,使数据更集中,以提高效率
  • repartition,是重新分区,增加或减少分区数量,数据随机重新分配,可以消除分区间的数据量差异
  • pipe,是与外部程序管道关联,从外部程序中获取数据。

Scala语法

在 RDD.scala中,几乎每一个转换和操作函数都会有一个withScope,例如:

def map[U: ClassTag](f: T => U): RDD[U] = withScope {
    val cleanF = sc.clean(f)
    new MapPartitionsRDD[U, T](this, (context, pid, iter) => iter.map(cleanF))
}
def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U] = withScope {
    val cleanF = sc.clean(f)
    new MapPartitionsRDD[U, T](this, (context, pid, iter) => iter.flatMap(cleanF))
}

withScope是一个函数,调用了RDDOperationScope.withScope方法:

private[spark] def withScope[U](body: => U): U = RDDOperationScope.withScope[U](sc)(body)

withScope就像是一个 AOP(面向切面编程),嵌入到所有RDD 的转换和操作的函数中,RDDOperationScope会把调用栈记录下来,用于绘制Spark UI的 DAG(有向无环图,可以理解为 Spark 的执行计划)。

我们用下面的代码简单演示一下 Scala 用函数做 AOP:

object Day1 {
  def main(args: Array[String]) = {
    Range(1,5).foreach(twice)
    println()
    Array("China", "Beijing", "HelloWorld").foreach(length)
  }

  def twice(i: Int): Int = aopPrint {
    i * 2
  }

  def length(s: String): Int = aopPrint {
    s.length
  }

  def aopPrint[U](i: => U): U = {
    print(i + " ")
    i
  }
}

aopPrint的 入参是“一个返回类型为U的函数”。这段程序中aopPrint就是一个模拟的切面,作用是把所有的函数返回值打印出来。结果是:

2 4 6 8 
5 7 10

从代码上看,aopPrint并没有降低代码的可读性。读者依然能很清楚地读懂twice和length函数。打印返回结果这个流程是独立于函数之外的切面。

结论

  1. RDD 的转换分图上几种
  2. RDD 的转换可以看成是产生新的 RDD,而新的 RDD 记录了每一个分区依赖上游的哪些分区、每个分区如何用上游分区计算而来
标签: spark 大数据 RDD

本文转载自: https://blog.csdn.net/Flychuer/article/details/134974625
版权归原作者 猿界零零七 所有, 如有侵权,请联系我们删除。

“Spark RDD的转换”的评论:

还没有评论