【Spark】Spark 基本概念、模块和架构

本文结构如下：

• 基本概念
• 基本模块
• 基本架构

一、基本概念

• RDD(Resilient Distributed Datasets)：弹性分布式数据集，只读分区记录的集合，Spark 对所处理数据的基本抽象。RDD 是 Spark 分发数据和计算的基础抽象类。一个 RDD 是一个不可改变的分布式集合对象，因此在使用 scala 编写时，前面加修饰符 val 。Spark 中 的计算可以简单抽象为对 RDD 的创建、转换和返回操作结果的过程：
  • 创建
    通过加载外部物理存储（如HDFS）中的数据集，或 Application 中定义的对象集合（如List）来创建。RDD 在创建后不可被改变，只可以对其执行下面两种操作。
  • 转换（Transformation）
    对已有的 RDD 中的数据执行计算进行转换，而产生新的 RDD，在这个过程中有时会产生中间 RDD。Spark 对于 Transformation 采用惰性计算机制，遇到 Transformation 时并不会立即计算结果，而是要等遇到 Action 时一起执行。
  • 行动（Action）
    对已有的 RDD 中的数据执行计算产生结果，将结果返回 Driver 程序或写入到外部物理存储。在 Action 过程中同样有可能生成中间 RDD。
• DAG(Directed Acyclic Graph)：有向无环图。在图论中，边没有方向的图称为无向图，如果边有方向称为有向图。在无向图的基础上，任何顶点都无法经过若干条边回到该点，则这个图就没有环路，称为有向无环图( DAG 图)。Spark 中使用 DAG 对 RDD 的关系进行建模，描述了 RDD 的依赖关系，这种关系也被称之为 lineage。
• Partition：分区。一个 RDD 在物理上被切分为多个 Partition，即数据分区，这些 Partition 可以分布在不同的节点上。Partition 是 Spark 计算任务的基本处理单位，决定了并行计算的粒度，而 Partition 中的每一条 Record 为基本处理对象。例如对某个 RDD 进行 map 操作，在具体执行时是由多个并行的 Task 对各自分区的每一条记录进行 map 映射。

• NarrowDependency：窄依赖。一个父 RDD 的 partition 最多被子 RDD 中的 partition 使用一次，一父对应一子。NarrowDependency 分为 OneToOneDependency 和 RangeDependency。

  
  image

• ShuffleDependency(WideDependency)：宽依赖。父 RDD 中的一个 partition 会被子 RDD 中的 partition 使用多次，一父多子。

  
  image
• Job：包含很多 task 的并行计算，可以认为是 Spark RDD 里面的 action，每个 action 的触发会生成一个 job。Spark 采用惰性机制，对 RDD 的创建和转换并不会立即执行，只有在遇到第一个 Action 时才会生成一个 Job，然后统一调度执行。一个 Job 包含 N 个 Transformation 和 1 个 Action。
• Shuffle：有一部分 Transformation 或 Action 会让 RDD 产生宽依赖，这样过程就像是将父 RDD 中所有分区的 Record 进行了“洗牌”（Shuffle），数据被打散重组，如属于 Transformation 操作的 join，以及属于 Action 操作的 reduce 等，都会产生 Shuffle。
• Stage：用户提交的计算任务是一个由 RDD 构成的 DAG，如果 RDD 在转换的时候需要做 Shuffle，那么这个 Shuffle 的过程就将这个 DAG 分为了不同的阶段（即Stage）。由于 Shuffle 的存在，不同的Stage 是不能并行计算的，因为后面 Stage 的计算需要前面 Stage 的 Shuffle 的结果。在对 Job 中的所有操作划分 Stage 时，一般会按照倒序进行，即从 Action 开始，遇到窄依赖操作，则划分到同一个执行阶段，遇到宽依赖操作，则划分一个新的执行阶段，且新的阶段为之前阶段的 parent，然后依次类推递归执行。child Stage 需要等待所有的 parent Stage执行完之后才可以执行，这时Stage 之间根据依赖关系构成了一个大粒度的 DAG 。在一个 Stage 内，所有的操作以串行的 Pipeline 的方式，由一组 Task 完成计算。
• Task：具体执行任务。一个 Job 在每个 Stage 内都会按照 RDD 的 Partition 数量，创建多个 task。每个 Stage 内多个并发的 Task 执行逻辑完全相同，只是作用于不同的Partition。一个 Stage 的总 Task 的个数由 Stage 中最后的一个 RDD 的 Partition 的个数决定。 在 Spark 中有两类 task:
  • ShuffleMapTask：输出是 shuffle 所需数据， stage 的划分也以此为依据， shuffle 之前的所有变换是一个 stage，shuffle 之后的操作是另一个 stage 。
  • ResultTask：输出是 result，比如 rdd.parallize(1 to 10).foreach(println) 这个操作没有 shuffle，直接就输出了，那么它的 task 是 resultTask，stage 也只有一个；如果是 rdd.map(x => (x, 1)).reduceByKey(_ + _).foreach(println)， 这个 job 因为有 reduce，所以有一个 shuffle 过程，那么 reduceByKey 之前的是一个 stage，执行 ShuffleMapTask，输出 shuffle 所需的数据，reduceByKey 到最后是一个 stage，直接就输出结果了。如果 job 中有多次 shuffle，那么每个 shuffle 之前都是一个stage。

附：spark中job stage task关系

二、基本模块

整个 Spark 主要由以下模块组成：

• Spark Core：Spark的核心功能实现，包括：
  • 基础设施：Spark 中有很多基础设施，被 Spark 中的各种组件广泛使用，这些基础设施包括 SparkConf、Spark 内置 RPC 框架、事件总线 ListenerBus、度量系统。
    • SparkConf 用于管理Spark应用程序的各种配置信息。
    • Spark 内置 RPC 框架 使用 Netty 实现，有同步和异步的多种实现，Spark 各个组件间的通信都依赖于此 RPC 框架。
    • 事件总线是 SparkContext 内部各个组件间使用事件——监听器模式异步调用的实现。
    • 度量系统由Spark中的多种度量源（Source）和多种度量输出（Sink）构成，完成对整个Spark集群中各个组件运行期状态的监控。
  • SparkContext：SparkContext 是 Spark 的入口，Spark 程序的提交与执行离不开 SparkContext。它隐藏了网络通信、分布式部署、消息通信、存储体系、计算引擎、度量系统、文件服务、Web UI 等内容，开发者只需要使用 SparkContext 提供的 API 完成功能开发。
  • SparkEnv：Spark 执行环境。SparkEnv 内部封装了 RPC 环境（RpcEnv）、序列化管理器、广播管理器（BroadcastManager）、map任务输出跟踪器（MapOutputTracker）、存储体系、度量系统（MetricsSystem）、输出提交协调器（OutputCommitCoordinator）等Task运行所需的各种组件。
  • 存储体系：它优先考虑使用各节点的内存作为存储，当内存不足时才会考虑使用磁盘，这极大地减少了磁盘 I/O，提升了任务执行的效率，使得 Spark 适用于实时计算、迭代计算、流式计算等场景。Spark 的内存存储空间和执行存储空间之间的边界是可以控制的。
  • 调度系统：调度系统主要由 DAGScheduler 和 TaskScheduler 组成。DAGScheduler 负责创建 Job、将 DAG 中的 RDD 划分到不同的 Stage、给 Stage 创建对应的 Task、批量提交 Task 等功能。TaskScheduler 负责按照 FIFO 或者 FAIR 等调度算法对批量 Task 进行调度。
  • 计算引擎等：计算引擎由内存管理器、任务内存管理器、Task、Shuffle 管理器等组成。
• Spark SQL：提供SQL处理能力，便于熟悉关系型数据库操作的工程师进行交互查询。此外，还为熟悉 Hive 开发的用户提供了对 Hive SQL 的支持。
• Spark Streaming：提供流式计算处理能力，目前支持 Apache Kafka、Apache Flume、Amazon Kinesis 和简单的 TCP 套接字等多种数据源。此外，Spark Streaming 还提供窗口操作用于对一定周期内的流数据进行处理。
• GraphX：提供图计算处理能力，支持分布式，Pregel 提供的 API 可以解决图计算中的常见问题。
• MLlib：Spark 提供的机器学习库。MLlib 提供了机器学习相关的统计、分类、回归等领域的多种算法实现。其一致的 API 接口大大降低了用户的学习成本。

Spark SQL、Spark Streaming、GraphX、MLlib的能力都是建立在核心引擎之上：

三、基本架构

Spark 集群由集群管理器 Cluster Manager、工作节点 Worker、执行器 Executor、驱动器 Driver、应用程序 Application 等部分组成。

3.1、Cluter Manager

Spark 的集群管理器，主要负责对整个集群资源的分配和管理。根据部署模式的不同，可以分为如下：

• Hadoop YARN: 主要是指 YARN 中的 ResourceManager。YARN 是 Hadoop2.0 中引入的集群管理器，可以让多种数据处理框架运行在一个共享的资源池上，让 Spark 运行在配置了 YARN 的集群上是一个非常好的选择，可以利用 YARN 来管理资源。
• Apache Mesos：主要是指 Mesos Master。Mesos 起源于Berkeley AMP实验室，是一个通用的集群管理器。能够将CPU、内存、存储以及其它计算资源由设备（物理或虚拟）中抽象出来，形成一个池的逻辑概念，从而实现高容错与弹性分布式系统的轻松构建与高效运行。
• Standalone：主要是指 Standalone Master。Standalone Master 是 spark 原生的资源管理，由Master负责资源的分配。

3.2、Worker

Spark 的工作节点，用于执行提交的作业。在 YARN 部署模式下 Worker 由 NodeManager 代替。

Worker 有如下作用：

• 通过注册机制向 Cluster Master 汇报自身的 cpu 和 memory 等资源
• 在 Master 的指示下创建启动 Executor，Executor 是执行真正计算的苦力
• 将资源和任务进一步分配给 Executor
• 同步资源信息、Executor 状态信息给 Cluster Master

3.3、Executor

真正执行计算任务的组件。

Executor 是某个 Application 运行在 Worker 节点上的一个进程，该进程负责运行某些 Task， 并且负责将数据存到内存或磁盘上，每个 Application 都有各自独立的一批 Executor， 在 Spark on Yarn 模式下，其进程名称为 CoarseGrainedExecutor Backend。一个 CoarseGrainedExecutor Backend 有且仅有一个 Executor 对象， 负责将 Task 包装成 taskRunner，并从线程池中抽取一个空闲线程运行 Task， 每个 CoarseGrainedExecutorBackend 能并行运行 Task 的数量取决于分配给它的 CPU 的个数。

3.4、Driver

Application 的驱动程序。可以理解为使程序运行中的 main 函数，它会创建 SparkContext。Application 通过 Driver 与 Cluster Master 和 Executor 进行通信。Driver 可以运行在 Application 中，也可以由 Application 提交给 Cluster Master，由 Cluster Master 安排 Worker 运行。

Driver 的作用：

• 运行应用程序的 main 函数
• 创建 Spark 的上下文
• 划分 RDD 并生成有向无环图（DAGScheduler）
• 与 Spark 中的其他组进行协调，协调资源等等（SchedulerBackend）
• 生成并发送 task 到 executor（taskScheduler）

3.5、Application

用户使用 Spark API 编写的的应用程序，其中包括一个 Driver 功能的代码和分布在集群中多个节点上运行的 Executor 代码。

Application 通过 Spark API 创建 RDD，对 RDD 进行转换，创建 DAG，并通过 Driver 将 Application 注册到 Cluster Master。