欢迎来到尧图网

客户服务 关于我们

您的位置:首页 > 健康 > 养生 > Spark--如何理解RDD

Spark--如何理解RDD

2025/2/6 16:10:15 来源:https://blog.csdn.net/2301_77948840/article/details/145456139  浏览:    关键词:Spark--如何理解RDD

1、概念

  rdd是对数据集的逻辑表示,本身并不存储数据,只是封装了计算逻辑,并构建执行计划,通过保存血缘关系来记录rdd的执行过程和历史(当一个rdd需要重算时,系统会根据血缘关系追溯到最初的数据源,重建丢失的数据),调用行动算子时,会从数据源读取数据并进行计算。

2、五大属性

(1)compute计算函数

描述本RDD的数据如何被计算出来,本质上是运算逻辑的迭代器。

(2)依赖RDD列表

一个或多个前序RDD。

(3)分区列表

RDD被分成多个分区。

(4)(k, v)类型rdd的分区器

  • 普通RDD:没有分区器,分区数在创建和Transformation时决定,后续可以通过repartition或coalesce修改。
  • PairRDD:具有分区器的概念,可以基于键分区,常用于需要快速聚合的场景。

(5)每个分区的首选计算执行位置

  为了提高计算效率,会根据数据本地化级别,将任务分配到离数据最近的计算节点进行计算。

3、本质

  rdd的本质是迭代器。

  迭代器是一种用于访问集合元素的设计模式,允许我们按需逐个访问集合中的元素,而无需一次性加载整个集合,允许一次仅访问一个元素,访问后可以前进到下一个元素,但无法返回上一个元素。
  RDD在调用行动算子(如collect,count,reduce等)时,每个Task中会创建个独立的迭代器。

执行具体过程:
(1)分区:当使用RDD时,数据被分成多个分区,每个分区可以独立处理。
(2)任务调度:当行动算子被调用时,spark会为每个分区创建一个任务(Task)。
(3)创建迭代器:在每个Task开始执行时,Spark会为该分区的RDD创建一个迭代器,从而能够逐个访问该分区的数据。
(4)逐个处理:迭代器以惰性方式逐一处理元素,执行你所定义的操作;例如,映射、过滤、聚合等。
(5)结果汇总:在所有分区的Task完成后,Spark将结果汇总,并返回给驱动程序。

  并行处理的优点:
(1)内存效率:每个Task只在内存中处理当前迭代器的数据,避免了同时加载整个RDD所需的数据。
(2)并行处理:每个Task可以在不同的Executor上并行执行,加快计速度。
(3)故障恢复:由于RDD的分区和迭代器的特性,Spark可以轻松地重算丢失的分区数据。

4、特点

(1)不可变性

一旦创建,rdd的内容就不能被修改了,可以通过转化操作创建一个新的rdd。

(2)弹性

可以在任务失败或数据丢失时,自动重算。

(3)支持分布计算

可以在整个集群中分布式地进行计算,支持大规模数据的处理。

5、RDD,DataFrame与DataSet的区别与联系

(1)RDD与DataFrame的区别

  RDD中的数据没有结构信息,是一种基础的数据结构,主要使用函数式编程风格。
  DataFrame是在RDD的基础上加上了一层schema,类似于表格的数据结构,有列名和数据类型的信息,提供了更加简洁的代码书写方式,支持SQL查询。


  上图直观地体现了DataFrame和RDD的区别。左侧的RDD[Person]虽然以Person为类型参数,但Spark框架本身不了解Person类的内部结构。而右侧的DataFrame却提供了详细的结构信息,使得Spark SQL可以清楚地知道该数据集中包含哪些列,每列的名称和类型各是什么。DataFrame多了数据的结构信息,即schema。RDD是分布式的Java对象的集合。DataFrame是分布式的Row对象的集合。DataFrame除了提供了比RDD更丰富的算子以外,更重要的特点是提升执行效率、减少数据读取以及执行计划的优化,比如filter下推、裁剪等。
RDD:

val rdd = sc.parallelize(Seq(1, 2, 3, 4, 5))
val result = rdd.map(_ * 2).collect()

DataFrame:

val df = spark.read.json("path/to/json")
val result = df.filter($"age" > 20).select("name", "age")

(2)DataFrame与DataSet的区别
DataFrame 可以看作是 DataSet[Row],其中 Row 是一个通用的行类型。DataFrame可以认为是Dataset的一个特例,主要区别是DataSet每一个record存储的是一个强类型值而不是一个Row。
DataSet:

case class Person(name: String, age: Int)
val ds = spark.read.json("path/to/json").as[Person]
val result = ds.filter(_.age > 20).map(_.name)

6、宽RDD和窄RDD的概念

RDD在计算过程中,会被划分成多个Stage,这依靠的就是RDD之间的依赖关系。RDD有2种依赖关系(宽依赖和窄依赖),根据不同的依赖关系来确定是否需要shuffle,根据是否需要shuffle来确定是否需要划分stage。

(1)窄依赖(NarrowDependency)有如下两种:

① OneToOneDependency
父RDD的分区与子RDD的分区是一一对应的关系。
② RangeDependency
父RDD与子RDD是多对一的关系,但是父RDD的分区与子RDD的分区是一对一的关系,所以分区之间并不会交叉,每个子RDD依然对应父RDD的一个分区。
在这里插入图片描述
窄RDD分区间的计算是一对一的,每个子RDD只需要读取父RDD的一个分区即可进行计算,所以不需要shuffle,即不需要划分stage。

2、宽依赖:

ShuffleDependency
在这里插入图片描述
ShuffleDependency中,每个子RDD的每个分区,都要拿到父RDD的每个分区的数据,才能进行计算。正因如此,在遇到宽依赖时,需要对数据进行shuffle处理,划分stage。

版权声明:

本网仅为发布的内容提供存储空间,不对发表、转载的内容提供任何形式的保证。凡本网注明“来源:XXX网络”的作品,均转载自其它媒体,著作权归作者所有,商业转载请联系作者获得授权,非商业转载请注明出处。

我们尊重并感谢每一位作者,均已注明文章来源和作者。如因作品内容、版权或其它问题,请及时与我们联系,联系邮箱:809451989@qq.com,投稿邮箱:809451989@qq.com