Spark RDD详解：从基本概念到执行机制

Spark RDD详解：从基本概念到执行机制

引用

简书

1.

https://www.jianshu.com/p/2c48b2f1c7f7?utm_campaign=maleskine&utm_content=note&utm_medium=seo_notes&utm_source=recommendation

Apache Spark中的RDD（弹性分布式数据集）是其核心数据结构，理解RDD对于掌握Spark框架至关重要。本文将从RDD的基本概念出发，深入探讨其主要属性、组成结构、分区机制、逻辑执行计划（Lineage）、依赖关系以及与Task/Stage的关系，帮助读者全面理解这一重要概念。

什么是RDD

RDD(Resilient Distributed Datasets)，弹性分布式数据集，是Spark的基本数据结构。它是一个不可变的分布式对象集合。RDD中的每个数据集被划分为逻辑分区，其可以在集群的不同节点上计算。RDD可以包含任何类型的Python，Java或Scala对象，包括用户定义的类。形式上，RDD是只读的分区记录集合。 可以通过读取外部存储系统中的数据集(如HDFS，HBase或提供Hadoop输入格式的任何数据源等)、转换现有数据集合或对其他RDD的数据进行转换来创建RDD。RDD是一个支持容错集合，可以并行操作。

RDD的主要属性

从RDD的内部定义来看，每个RDD拥有以下五个主要属性：

• 分区列表
• 与其他RDD的依赖关系列表
• 计算分片(split)的函数
• (可选) 键值RDD中的分区器Partitioner (例如，hash-partitioner)
• (可选) 用于计算每个分片的的优选位置列表 (例如，HDFS文件的block位置)

RDD的组成

RDD主要由以下四部分组成

• 分区(Partitions)：数据集的原子片段。 每个计算节点含有一个或多个分区。
• 依赖关系(Dependencies)：RDD的每个分区计算时依赖哪些父RDD的分区(如下图)
• 函数/算子(Functions): 基于其父RDD的用于计算数据集的函数。
• 元数据(Metadata): RDD的分区方案和数据的存储放置。
  
  Credit: https://github.com/rohgar/scala-spark-4/wiki/Wide-vs-Narrow-Dependencies
  
  Credit: https://github.com/rohgar/scala-spark-4/wiki/Wide-vs-Narrow-Dependencies

RDD的分区(Partition)

RDD中的数据被存储在多个分区中。
RDD分区的特征

• 分区永远不会跨越多台机器，即同一分区中的数据始终保证在同一台机器上。
• 群集中的每个节点包含一个或多个分区。
• 分区的数目是可以设置的。 默认情况下，它等于所有执行程序节点上的核心总数。 例如。 6个工作节点，每个具有4个核心，RDD将被划分为24个分区。
  RDD分区与任务执行的关系
  在Map阶段partition数目保持不变。
  在Reduce阶段，RDD的聚合会触发shuffle操作，聚合后的RDD的partition数目跟具体操作有关，例如repartition操作会聚合成指定分区数，还有一些算子是可配置的。
  RDD分区数的调整可以通过以下两个函数完成：
• repartition
  repartition函数相当于coalesce(numPartitions, shuffle =True), 不仅可以调整分区数目(增加或减少)，也可以将partitioner调整为hash-partitioner，产生shuffle操作
• coalesce
  coalesce函数可以控制是否shuffle，但是shuffle为False时，只能减少分区数，无法增大。
  RDD在计算的时候，每个分区都会启动一个task，RDD的分区数目决定了总的task数目。
  申请的Executor数和Executor的CPU核数，决定了你同一时刻可以并行执行的task数量。
  这里我们举个例子来加深对RDD分区数量与task执行的关系的理解
  比如的RDD有100个分区，那么计算的时候就会生成100个task，你的资源配置为10个计算节点，每个两2个核，同一时刻可以并行的task数目为20，计算这个RDD就需要5个轮次。如果计算资源不变，你有101个task的话，就需要6个轮次，在最后一轮中，只有一个task在执行，其余核都在空转。
  partition数量太少会造成资源利用不够充分。
  例如，在资源不变的情况，你的RDD只有10个分区，那么同一时刻只有10个task运行，其余10个核将空转。
  通常在spark调优中，可以增大RDD分区数目来增大任务并行度。
  但是partition数量太多则会造成task过多，task的传输/序列化开销增大，也可能会造成输出过多的(小)文件。
  <spark.default.parallelism 和 spark.sql.shuffle.partitions 这两个参数很重要
  RDD的分区器(Partitioner)
  Spark中提供两种分区器：
• 散列分区 Hash partitioning
• 范围分区 Range partitioning
  只有PairRDD支持自定义分区器。