Spark学习笔记

# 一、Spark概述

## Spark 风雨十年

### 1.1 十年发展

加州伯克利分校AMP实验室开发的通用大数据处理框架。

![image-20220903121109356](http://kmgy.top:9090/image/2022/9/3/image-20220903121109356_repeat_1662178273065__392842.png)

### 1.2 Spark 风雨十年

Spark从2016年开始后开源问题提交量最多的项目。

![image-20220903122723773](http://kmgy.top:9090/image/2022/9/3/image-20220903122723773_repeat_1662179243940__490036.png)

## Sparak和Hadoop区别

### 1.1 技术栈区别

|              | Hadoop                                    | Spark                                                      |
| ------------ | ----------------------------------------- | ---------------------------------------------------------- |
| 类型         | 基础平台，包含计算，存储，调度            | 纯计算工具                                                 |
| 场景         | 海量数据批处理（磁盘迭代计算）            | 海量数据批处理（内存迭代几段、交互式计算）、海量数据流计算 |
| 价格         | 便宜                                      | 对内存有要求，价格高                                       |
| 编程范式     | MR,API较为底层，算法适应性差              | RDD组成DAG有向无环图，API较为顶层，方便使用                |
| 数据存储结构 | MapReduce中间计算结果在HDFS磁盘上，延迟大 | RDD中间运算结果在内存中，延迟小                            |
| 运行方式     | Task以进程方式维护，任务启动慢            | Task以线程方式维护，任务启动快，可批量创建提高并行能力。   |

## Spark 4大特点

### 特点

- 速度快
  - Spark比MR快100倍，硬盘快10倍
    - 1.可以将结果放在内存中处理
    - 2.提供了丰富的API算子，做到在一个Spark程序中完成
- 易于使用
- 通用性强
  - Spark 提供了Spark SQL、Spark Streaming 、MLib及GraphX在内的多个工具库，我们可以在一个应用中无缝地使用这些工具库。
  - ![image-20220903135147353](http://kmgy.top:9090/image/2022/9/3/image-20220903135147353_repeat_1662184307963__042022.png)
- 运行方式
  - Spark支持多种运行方式，包括在Hasoop和Mesos上，也支持Standalone的独立运行模式，同时也可以运行在K8S上。
  - Sprak可以读取多种数据源
  - ![image-20220903135518793](http://kmgy.top:9090/image/2022/9/3/image-20220903135518793_repeat_1662184518997__195987.png)

## Spark 框架模块

### 模块组成

- Spark Core
  - Spark的核心，提供结构化数据处理模块。以RDD为数据抽象提供各种语言API，实现海量离线数据批处理计算。
- Spark SQL
  - 基于Core，支持以Core对数据进行处理，针对的是离线场景。Spark提供了StructuredStreaming，可以用SparkSQL为基础进行流式计算。
- Spark Streaming
  - 基于Core为基础，提供流式计算（诞生较早。有一定缺陷）
- Spark GraphX
  - 图计算
- Spark MLib
  - 机器学习模块

## Spark 运行模式

### 运行模式

![image-20220903140838277](http://kmgy.top:9090/image/2022/9/3/image-20220903140838277_repeat_1662185318459__465906.png)

- 本地模式（Local 开发和测试）
  - 以一个独立的进程，通过内部的多个线程来模拟整个Spark运行时环境。
- Standalone（集群）
  - 各个角色独立进程形式的存在，并组成Spark集群环境。
- Hadoop Yarn模式
  - 各个角色运行在Yarn容器内部，并组成Spark集群环境。
- K8S（容器集群）
  - 各个角色运行在K8S容器内部，并组成Spark集群环境
- 云服务模式（运行在云平台上）

## Spark 节点角色

![image-20220903141439654](http://kmgy.top:9090/image/2022/9/3/image-20220903141439654_repeat_1662185679837__111626.png)

### 资源层面

- Master角色：集群资源管理
- Worker的角色：单机资源管理

### Worker中任务运行层面

- Driver：单个任务的管理者

- Executor: 任务运行功能

## 总结

Spark 解决了海量数据的计算，可以进行理想批处理以及实时流计算。

Spark 5大模块

Spark 4大特点

Spark 3大运行模式

Spark 4种运行角色

# 二、Spark部署

## Spark Local模式

![image-20220904095541173](http://kmgy.top:9090/image/2022/9/4/image-20220904095541173_repeat_1662256541301__117423.png)

### Local模式下角色分布：

资源管理：

- Master：Local进程本身

- Worker：Local进程本身

任务执行：

- Driver：Local进程本身
- Executor：不存在，没有独立的Executor角色，由Local进程内的线程提供计算能力

**注意：Driver也是一直特殊的Executor，只不过多数时候，我们将Executor当做纯Worker来对待。**

![image-20220904100015803](http://kmgy.top:9090/image/2022/9/4/image-20220904100015803_repeat_1662256815898__073951.png)

## Spark Standalone 模式

![image-20220904100518841](http://kmgy.top:9090/image/2022/9/4/image-20220904100518841_repeat_1662257119031__197795.png)

StandAlone是真实地在多个机器之间搭建Spark集群的环境，完全可以利用该模式搭建多个机器集群，用于实际的大数据处理。

### Standalone模式下角色分布：

- 主节点
  - Master角色，管理整个集群资源，并托管运行各个任务的Driver
- 从节点Workers
  - 管理每个极其的资源，分配对应的资源来运行Executor
  - 每个从节点分配资源信息给Worker，资源信息包含内存Memory和CPU Cores核数
- 历史服务器HistorySever(可选)
  - 保存事件日志数据至HDFS，启动HistoryServer可以查看应用运行和相关信息
- ![image-20220904183429392](http://kmgy.top:9090/image/2022/9/4/image-20220904183429392_repeat_1662287670639__339802.png)

## 其他

### 1. RDD:

#### 1.2.1 RDD概念

![Spark 理论基石 —— RDD](http://kmgy.top:9090/image/2022/9/3/v2-f20bd6bce4094ea7800dbcf1ed56504d_1440w.jpg_repeat_1662179793710__715449.jpeg)

- RDD学名可伸缩的分布式数据集（Resilient Distributed Dataset）。是一种对数据集形态的抽象，基于此抽象，使用者可以在集群中执行一系列计算，而不用将中间结果落盘。
- 对于分布式系统，容错支持是必不可少的。为了支持容错，RDD 只支持粗粒度的变换。即，输入数据集是 immutable （或者说只读）的，每次运算会产生新的输出。不支持对一个数据集中细粒度的更新操作。这种约束，大大简化了容错支持，并且能满足很大一类的计算需求。
- RDD 是一个基于分区的、只读的数据记录集抽象。RDD 只可以通过对持久存储或其他 RDD 进行确定性运算得来，这种运算被称为**变换**。常用的**变换**算子包括：map，filter 和 join。
- RDD 没有选择不断的做检查点以进行容错，而是会记下 RDD 从最初的外存的数据集变化而来的变化路径，也就是其**谱系**（lineage）。理论上所有的 RDD 都可以在出错后从外存中依据谱系图进行重建。一般来说，重建的粒度是分区（Partition）而非整个数据集，一来代价更小，二来不同分区可能在不同机器上。
- **用户可以对 RDD 的两个方面进行控制：持久化和分区控制。**对于前者，如果某些 RDD 需要复用，那么用户可以指示系统按照某种策略将其进行持久化。后者来说，用户可以定制分区路由函数，将数据集合中的记录按照某个键值路由到不同分区。比如进行 Join 操作的时候，可以讲待 Join 数据集按照相同的策略进行分区，以并行 Join。
- **数据落盘**
  - 在很多的场景中，我们经常要确保数据已经安全的写到磁盘，以便在系统宕机或重启之后还能读到这些数据。但是我们知道，linux系统的IO路径很复杂，并且分为很多层，每一层可能都会有buffer来加速IO读写。因此，想要将数据安全的写到磁盘，并不是简单调一个write/fwrite就可以搞定的。

#### 1.2.2 举例

假设我们利用 Spark 接口相对存在于 HDFS 上的日志文件，找出错误条目，针对出现 hdfs 关键字的具体条目进行分析。

```scala
//基于某个 hdfs 上的文件定义一个 rdd（每一行作为集合中的一个条目）
lines = spark.textFile("hdfs://...") 
//第二行通过 filter 变换生成新的 rdd
errors = lines.filter(_.startsWith("ERROR"))
//第三行请求 spark 将其结果进行暂存
errors.persist()

// 以一个 collect 的动作结尾，求出包含 HDFS 关键字的所有行数的各个字段。
//(assuming time is field
// number 3 in a tab-separated format):
errors.filter(_.contains("HDFS"))
      .map(_.split(’\t’)(3))
      .collect()
```

- 第一行基于某个 hdfs 上的文件定义一个 rdd（每一行作为集合中的一个条目）。
- 第二行通过 filter 变换生成新的 rdd
- 第三行请求 spark 将其结果进行暂存
- 最后一行以一个 collect 的动作结尾，求出包含 HDFS 关键字的所有行数的各个字段。

其计算谱系图如下：

![img](http://kmgy.top:9090/image/2022/9/3/v2-7f65c0b745407804523f4df6bd14fb58_b.jpg_repeat_1662180709217__667273.jpeg)

有两点需要注意：

1. **直到遇到 collect 这个动作（Action）之前都没有发生实际的运算。**
2. **链式操作时不保存中间结果；**

由于第三行将结果在内存中进行了缓存，因此还可以基于此做其他动作。比如，计算包含 'MySQL' 关键字的错误条数：

```scala
// Count errors mentioning MySQL:
errors.filter(_.contains("MySQL")).count()
```

#### 1.2.3 **RDD 模型的优点**

1. 使用 lineage 来按需恢复数据，而不用定期 snapshot，减小了不必要开销。
2. 每个 Partition 出错后可以单独进行恢复，而不用进行全数据集的重建。
3. RDD 舍弃了对任意内存位置进行更新，只允许批量的写入数据，从而提高了容错效率
4. RDD 的不可变的特点允许系统叫较容易的对某些计算进行迁移，不用考虑一致性的问题。
5. 由于只支持批量计算，因此调度系统可以比较好的利用数据局部性的特点加快运算速度。
6. 由于只支持批量计算，因此调度系统可以比较好的利用数据局部性的特点加快运算速度。

#### 1.2.4 Spark编程接口

- Spark 通过暴露与编程语言集成的算子来提供操作 RDD 的接口。 **其中 RDD 表现为编程语言中的类，而 RDD 的算子为作用于这些类上的函数。**之前的系统如 DryadLINQ 和 FlumeJava 也使用了类似的形式。

- 用户使用 RDD 时，首先将数据从持久化存储中通过*变换*（**Transformations**，如 *map* 或者 *filter*）将其载入内存，然后可以对 RDD 施加任何系统支持的一系列变换，最后利用*动作*（**Action**）算子，将 RDD 重新持久化到外存中或者将控制权交还用户。

- 这个加载-变换-落盘的过程是**声明式**（Declarative，或者说是惰式[2]）的，Spark 在拿到整个拓扑后会利用执行引擎进行执行优化（比如将并行化、流水线化，之后会进一步讨论）。
- 此外很重要的一个接口是 *persist*，可以由用户来告诉系统**哪些** RDD 需要持久化，**如何**持久化（本机硬盘还是跨机器存储），如果有多个 RDD 需要持久化，那么**优先级**如何确定。Spark 默认将 RDD 保存在内存中，如果内存不够用了会根据用户配置将数据溢出（spill）到硬盘上。

- **声明式**
  - 它可以看作是命令式的反面。曾有人言：一切非命令式，皆是声明式。从这个意义上说，越是偏离图灵机的图像越远的，就越是声明式的。
  - 所以，**函数式编程（Functional Programming）是声明式的，因为它不使用可变状态，也不需要指定任何的执行顺序关系（**可以假定所有的函数都是同时执行的，因为存在引用透明性，所谓的参数和变量都只是一堆符号的别名而已）。逻**辑式编程（Logical Programming）也是声明式的，因为我们只需要通过facts和rules描述我们所需要解决的问题，具体的求解路径由编译器和程序运行时自动决定。**
- Spark 利用 Scala 语言作为 RDD 抽象的接口，因为 Scala 兼顾了精确（其函数式语义适合交互式场景）与高效（使用静态类型）。

- 开发者利用 Spark 提供的库编写*驱动程序* （driver programe）以使用 Spark。**驱动程序会定义一到多个 RDD，并对其进行各种变换。Spark 提供的库会连接 Spark 集群，生成计算拓扑，并将拓扑分散到多个 workers 上去进行执行，同时记下变换的谱系（lineage）。**这些 workers 是分散在 Spark 集群内各个机器上的常驻进程，它们在内存里保存计算过程中生成的 RDD 的各个分区。
- ![img](http://kmgy.top:9090/image/2022/9/3/v2-280401515a5506cd19592550ca434e4e_b.jpg_repeat_1662182083407__703476.jpeg)

#### 1.2.5 RDD的表示

RDD 抽象的核心组成主要有以下五个部分：

1. **分区集**（partition set）。分区是每个 RDD 的最小构成单元。
2. **依赖集**（dependencies set）。主要是 RDD 间的父子依赖关系。
3. **变换函数**（compute function）。作用于分区上的变换函数，可以由几个父分区计算得到一个子分区。
4. **分区模式**（partition scheme）。该 RDD 的分区是基于**哈希分片**的还是**直接切分**的。
5. **数据放置**（data placement）。知道分区的存放位置可以进行计算优化。

在 RDD 的接口设计中最有趣的一个点是如何对 RDD 间的**依赖关系**进行规约。最后发现可以将所有依赖归纳为两种类型：

1. **窄依赖**（narrow dependencies）：父 RDD 的分区最多被一个子 RDD 的分区所依赖，比如 *map*。
2. **宽依赖**（wide dependencies）：父 RDD 的分区可能被多个子 RDD 的分区所依赖，比如 *join*。

![preview](http://kmgy.top:9090/image/2022/9/3/v2-8ebc0b892f5de7ededd99eb561520816_r.jpg_repeat_1662183341861__717513.jpeg)

**调度优化**。对于窄依赖，可以对分区间进行并行**流水化**调度，先计完成某个窄依赖算子（比如说 map）的分区不用等待其他分区而直接进行下一个窄依赖算子（比如 filter ）的运算。与之相对，宽依赖的要求父 RDD 的所有分区就绪，并进行跨节点的传送后，才能进行计算。类似于 MapReduce 中的 shuffle。

**数据恢复**。在某个分区出现错误或者丢失时，窄依赖的恢复更为高效。因为涉及到的父分区相对较少，并且可以并行恢复。而对于宽依赖，由于依赖复杂（如上图，子 RDD 的每个分区都会依赖父 RDD 的所有分区），一个分区的丢失可能就会引起全盘的重新计算。

**HDFS 文件**：*partitions* 函数返回 HDFS 文件的所有 block，每个 block 被当做一个 partition。 *preferredLocations* 返回每个 block 所在的位置，*Iterator* 会对每个 block 进行读取。

**map**：在任意 RDD 上调用 map 会返回一个 MappedRDD 对象，该对象的 *partitions* 函数和 *preferredLocations* 与父 RDD 保持一致。对于 *iterator*，只需要将传给 map 算子的函数以此作用到其父 RDD 的各个分区即可。

**union**: 在两个 RDD 上调用 union 会返回一个新的 RDD，该 RDD 的每个分区由对应的两个父 RDD 通过窄依赖计算而来。

**sample**：抽样函数和 map 大体一致。但该函数会给每个分区保存一个随机数种子来决定父 RDD 的每个记录是否保留。

**join**：在两个 RDD 上调用 join 操作可能会导致两个窄依赖（比如其分区都是按待 join 的key 哈希的），两个宽依赖，或者混合依赖。每种情况下，子 RDD 都会有一个 *partitioner* 函数，或继承自父分区，或是默认的hash 分区函数。

##