SparkCore（一）：Spark简介和运行环境的搭建

左手的ㄟ右手 2022-12-01 15:56 156阅读 0赞

### 文章目录 ###

*  一、Spark简介
 *   *  1.1 Hadoop的发展
     *  1.2 Spark介绍
     *  1.3 Spark or Hadoop
     *  1.4 Spark 核心模块
     *  1.5 Spark快速上手
 *  二、Spark运行环境
 *   *  2.1 Local模式
     *  2.2 Standalone模式
     *  2.3 Yarn模式【重点】
     *  2.4 K8S & Mesos模式
     *  2.5 常用三种运行环境对比
 *  三、Spark运行架构
 *   *  3.1 运行架构
     *  3.2 核心组件
     *  3.3 核心概念
     *  3.4 提交流程

# 一、Spark简介 #

## 1.1 Hadoop的发展 ##

**Hadoop 1.x**

从架构的角度`Hadoop 1.X`存在很多的问题

1.  `Namenode`是单点操作，所以容易出现单点故障，制约了`HDFS`的发展；
2.  `Namenode`的内存限制也影响了`HDFS`的发展；
3.  `MapReduce`是一种基于数据集的工作模式，面向数据，这种工作模式一般是从存储上加载数据集，然后操作数据集，最后将结果写入物理存储设备。数据更多面临的是一次性计算，所以初衷是单一数据计算，不支持迭代计算；
4.  资源调度和任务调度耦合在一起，无法扩展，所以`Hadoop1.X`版本只支持`MR`计算框架。

**Hadoop 2.x**

1.  `2.x`版本支持`Namenode`高可用；
2.  `2.x`版本使用新的资源调度框架`Yarn`，只做资源调度，不进度；
3.  `MR`框架只做任务调度，可插拔，所以扩展性非常的强。

## 1.2 Spark介绍 ##

![在这里插入图片描述][20200917152353231.png]  
`Spark`是一种基于内存的快速、通用、可扩展的大数据分析计算引擎。

`Spark`最初由美国加州伯克利大学的`AMPLab`于2009年使用`Scala`开发，是基于内存计算的大数据并行计算框架，可用于构建大型的、低延迟的数据分析应用程序。

2013年6月，`Spark`成为了`Apache`基金会下的项目；同年`Hadoop`在2013年10月发布`2.X` (`Yarn`)版本。

## 1.3 Spark or Hadoop ##

`Hadoop`的`MR`框架和`Spark`框架都是数据处理框架，那么我们在使用时如何选择呢？

*  `Hadoop MapReduce`由于其设计初衷并不是为了满足循环迭代式数据流处理，因此在多并行运行的数据可复用场景（如：机器学习、图挖掘算法、交互式数据挖掘算法）中存在诸多计算效率等问题。所以`Spark`应运而生，`Spark`就是在传统的`MapReduce`计算框架的基础上，利用其计算过程的优化，从而大大加快了数据分析、挖掘的运行和读写速度，并将计算单元缩小到更适合并行计算和重复使用的`RDD`计算模型。
 *  机器学习中`ALS`、凸优化梯度下降等。这些都需要基于数据集或者数据集的衍生数据反复查询反复操作。`MR`这种模式不太合适，即使多`MR`串行处理，性能和时间也是一个问题。数据的共享依赖于磁盘。另外一种是交互式数据挖掘，`MR`显然不擅长。而`Spark`所基于的`scala`语言恰恰擅长函数的处理。
 *  `Spark`是一个分布式数据快速分析项目。它的核心技术是弹性分布式数据集（`Resilient Distributed Datasets`），提供了比`MapReduce`丰富的模型，可以快速在内存中对数据集进行多次迭代，来支持复杂的数据挖掘算法和图形计算算法。
 *  `Spark`和`Hadoop`的根本差异是多个作业之间的数据通信问题 : `Spark`多个作业之间数据通信是基于内存，而`Hadoop`是基于磁盘。
 *  `Spark Task`的启动时间快。`Spark`采用`fork`线程的方式，而`Hadoop`采用创建新的进程的方式。
 *  `Spark`只有在`shuffle`的时候将数据写入磁盘，而`Hadoop`中多个MR作业之间的数据交互都要依赖于磁盘交互
 *  `Spark`的缓存机制比`HDFS`的缓存机制高效。

经过上面的比较，我们可以看出在绝大多数的数据计算场景中，`Spark`确实会比`MapReduce`更有优势。但是`Spark`是基于内存的，所以在实际的生产环境中，由于内存的限制，可能会由于内存资源不够导致`Job`执行失败，此时`MapReduce`其实是一个更好的选择，所以`Spark`并不能完全替代`MR`。

## 1.4 Spark 核心模块 ##

![在这里插入图片描述][20200917145844473.png]

*  **Spark Core**：`Spark Core`中提供了`Spark`最基础与最核心的功能，`Spark`其他的功能如：`Spark SQL`，`Spark Streaming`，`GraphX`， `MLlib`都是在`Spark Core`的基础上进行扩展的
 *  **Spark SQL**：`Spark SQL`是`Spark`用来操作结构化数据的组件。通过`Spark SQL`，用户可以使用`SQL`或者`Apache Hive`版本的`SQL`方言（`HQL`）来查询数据。
 *  **Spark Streaming**：`Spark Streaming`是`Spark`平台上针对实时数据进行流式计算的组件，提供了丰富的处理数据流的`API`。
 *  **Spark MLlib**：`MLlib`是`Spark`提供的一个机器学习算法库。`MLlib`不仅提供了模型评估、数据导入等额外的功能，还提供了一些更底层的机器学习原语。
 *  **Spark GraphX**：`GraphX`是`Spark`面向图计算提供的框架与算法库。

## 1.5 Spark快速上手 ##

创建`Maven`项目并引入依赖：

<dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>2.4.5</version>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.2.2</version>
                <executions>
                    <execution>
                        
                        <goals>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>3.0.0</version>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

代码如下：

//1.准备Spark环境
        val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
        //2.建立和Spark的连接
        val sc = new SparkContext(sparkConf)
        //3.读取文件
        val file: RDD[String] = sc.textFile("input")
        //4.根据行拆分
        val words: RDD[String] = file.flatMap(line => line.split(" "))
        //5.根据key分组
        val wordCount: RDD[(String, Iterable[String])] = words.groupBy(word => word)
        //6.映射并排序
        val result = wordCount.map(t => (t._1, t._2.size)).sortBy(t => t._2).collect()
    
        /** 6.7步可缩短为以下一步 */
        //val result = words.map((_,1)).reduceByKey(_+_).sortBy(t => t._2).collect()
        //7.打印结果
        result.foreach(t => print(t + " ")) //(World,1) (Test,2) (Scala,2) (Spark,4) (Hello,5)
        //8.关闭连接
        sc.stop()

# 二、Spark运行环境 #

`Spark`作为一个数据处理框架和计算引擎，被设计在所有常见的集群环境中运行，在国内工作中主流的环境为`Yarn`，不过逐渐容器式环境也慢慢流行起来。接下来，我们就分别看看不同环境下`Spark`的运行：

![在这里插入图片描述][2020092215213139.png]

## 2.1 Local模式 ##

`Local`模式，就是不需要其他任何节点资源就可以在本地执行`Spark`代码的环境，一般用于教学，调试，演示等。它可以通过以下集中方式设置`Master`。

<table> 
 <thead> 
  <tr> 
   <th>local模式名称</th> 
   <th>效果</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>local</td> 
   <td>运行在一个线程</td> 
  </tr> 
  <tr> 
   <td>local[K]</td> 
   <td>运行在K个线程</td> 
  </tr> 
  <tr> 
   <td>local[*]</td> 
   <td>CPU的最大核值来设置线程数</td> 
  </tr> 
 </tbody> 
</table>

**搭建Local环境：**

① 安装`Spark`

[root@hadoop100 ~]# tar -zxvf spark-2.4.5-bin-hadoop2.7.tgz -C /opt/module/
    [root@hadoop100 ~]# mv spark-2.4.5-bin-hadoop2.7 spark-2.4.5

② 提交应用

[root@hadoop100 spark-2.4.5]# bin/spark-submit \
    --class org.apache.spark.examples.SparkPi \
    --master local[2] \
    ./examples/jars/spark-examples_2.11-2.4.5.jar \
    10

参数说明：

*  `--class`：表示要执行程序的主类
 *  `--master local[2]`：部署模式，默认为本地模式，数字表示分配的虚拟`CPU`核数量
 *  `spark-examples_2.12-2.4.5.jar`：运行的应用类所在的`jar`包
 *  `10`：表示程序的入口参数，用于设定当前应用的任务数量

## 2.2 Standalone模式 ##

`Local`本地模式毕竟只是用来进行练习演示的，真实工作中还是要将应用提交到对应的集群中去执行，这里我们来看看只使用`Spark`自身节点运行的集群模式，也就是我们所谓的独立部署（`Standalone`）模式。`Spark`的`Standalone`模式体现了经典的`master-slave`模式。

**搭建Standalone环境：**

① 修改`spark`配置文件

[root@hadoop100 conf]# vim slaves
    hadoop100
    hadoop101
    hadoop102
    [root@hadoop100 conf]# vim spark-env.sh
    SPARK_MASTER_HOST=hadoop100
    SPARK_MASTER_PORT=7077

② 分发`spark`包

[root@hadoop100 module]# xsync spark-2.1.1

③启动`spark`集群

[root@hadoop100 spark-2.1.1]# sbin/start-all.sh

④ 提交应用

[root@hadoop100 spark-2.1.1]# bin/spark-submit \
    --class org.apache.spark.examples.SparkPi \
    --master spark://hadoop100:7077 \
    ./examples/jars/spark-examples_2.12-2.4.5.jar \
    10

参数：`--master spark://hadoop100:7077`指定要连接的集群的`master`

## 2.3 Yarn模式【重点】 ##

`Standalone`模式由`Spark`自身提供计算资源，无需其他框架提供资源。这种方式降低了和其他第三方资源框架的耦合性，独立性非常强。但是`Spark`主要是计算框架，而不是资源调度框架，所以本身提供的资源调度并不是它的强项，所以`Yarn`环境下使用`Spark`尤为重要。  
![在这里插入图片描述][20200922160459721.png]  
① 修改`hadoop`配置文件`yarn-site.xml`，添加如下内容

<property>
    	<name>yarn.nodemanager.pmem-check-enabled</name>
    	<value>false</value>
    </property>
    
    <property>
    	<name>yarn.nodemanager.vmem-check-enabled</name>
    	<value>false</value>
    </property>

② 修改`spark-env.sh`，添加如下配置：

[root@hadoop100 conf]$ vi spark-env.sh
    YARN_CONF_DIR=/opt/module/hadoo-2.7.2/etc/hadoop

③ 分发配置文件

[root@hadoop100 conf]# xsync /opt/module/hadoop-2.7.2/etc/hadoop/yarn-site.xml

④ 提交应用

[root@hadoop100 spark-2.1.1]# bin/spark-submit \
    --class org.apache.spark.examples.SparkPi \
    --master yarn \
    ./examples/jars/spark-examples_2.12-2.4.5.jar \
    10

注意：在提交任务之前需启动`HDFS`以及`YARN`集群

## 2.4 K8S & Mesos模式 ##

`Mesos`是`Apache`下的开源分布式资源管理框架，它被称为是分布式系统的内核，在`Twitter`得到广泛使用，管理着`Twitter`超过300000台服务器上的应用部署，但是在国内，依然使用着传统的`Hadoop`大数据框架，所以国内使用`Mesos`框架的并不多，但是原理其实都差不多，这里我们就不做过多讲解了。  
![在这里插入图片描述][20200922155530323.png]  
容器化部署是目前业界很流行的一项技术，基于`Docker`镜像运行能够让用户更加方便地对应用进行管理和运维。容器管理工具中最为流行的就是`Kubernetes`（`K8S`），而`Spark`也在最近的版本中支持了`K8S`部署模式。这里我们也不做过多的讲解，可以参考[官网][Link 1]。  
![在这里插入图片描述][20200922155631730.png]

## 2.5 常用三种运行环境对比 ##

<table> 
 <thead> 
  <tr> 
   <th>模式</th> 
   <th>Spark安装机器数</th> 
   <th>需启动的进程</th> 
   <th>所属者</th> 
   <th>应用场景</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Local</td> 
   <td>1</td> 
   <td>无</td> 
   <td>Spark</td> 
   <td>测试</td> 
  </tr> 
  <tr> 
   <td>Standalone</td> 
   <td>3</td> 
   <td>Master及Worker</td> 
   <td>Spark</td> 
   <td>单独部署</td> 
  </tr> 
  <tr> 
   <td>Yarn</td> 
   <td>1</td> 
   <td>Yarn及HDFS</td> 
   <td>Hadoop</td> 
   <td>混合部署</td> 
  </tr> 
 </tbody> 
</table>

`Spark`常用端口号：

*  `Spark`查看当前`Spark-shell`运行任务情况端口号：4040（计算）
 *  `Spark Master`内部通信服务端口号：7077
 *  `Standalone`模式下，`Spark Master Web`端口号：8080（资源）
 *  `Spark`历史服务器端口号：18080
 *  `Hadoop YARN`任务运行情况查看端口号：8088

# 三、Spark运行架构 #

## 3.1 运行架构 ##

`Spark`框架的核心是一个计算引擎，整体来说，它采用了标准`master-slave`的结构。如下图所示，它展示了一个`Spark`执行时的基本结构。图形中的`Driver`表示`master`，负责管理整个集群中的作业任务调度。图形中的`Executor`则是`slave`，负责实际执行任务。  
![在这里插入图片描述][20200922150832259.png]

## 3.2 核心组件 ##

由上图可以看出，对于`Spark`框架有两个核心组件：  
![在这里插入图片描述][20200922160306871.png]  
**① Driver（驱动器）**

`Spark`驱动器节点，用于执行`Spark`任务中的`main`方法，负责实际代码的执行工作。`Driver`在`Spark`作业执行时主要负责：

主要负责：

*  把用户程序转为作业（`job`）
 *  在`Executor`之间调度任务(`task`)
 *  跟踪`Executor`的执行情况
 *  通过`UI`展示查询运行情况

实际上，我们无法准确地描述`Driver`的定义，因为在整个的编程过程中没有看到任何有关`Driver`的字眼。所以简单理解，所谓的`Driver`就是驱使整个应用运行起来的程序，也称之为`Driver`类。

**② Executor（执行器）**

`Spark Executor`是集群中工作节点（`Worker`）中的一个`JVM`进程，负责在`Spark`作业中运行具体任务（`Task`），任务彼此之间相互独立。`Spark`应用启动时，`Executor`节点被同时启动，并且始终伴随着整个`Spark`应用的生命周期而存在。如果有`Executor`节点发生了故障或崩溃，`Spark`应用也可以继续执行，会将出错节点上的任务调度到其他`Executor`节点上继续运行。

`Executor`有两个核心功能：

*  负责运行组成`Spark`应用的任务，并将结果返回给驱动器进程
 *  通过自身的块管理器（`Block Manager`）为用户程序中要求缓存的`RDD`提供内存式存储。`RDD`是直接缓存在`Executor`进程内的，因此任务可以在运行时充分利用缓存数据加速运算

**ApplicationMaster**

`Hadoop`用户向`YARN`集群提交应用程序时,提交程序中应该包含`ApplicationMaster`，用于向资源调度器申请执行任务的资源容器`Container`，运行用户自己的程序任务`job`，监控整个任务的执行，跟踪整个任务的状态，处理任务失败等异常情况。

说的简单点就是，`ResourceManager`（资源）和`Driver`（计算）之间的解耦合靠的就是`ApplicationMaster`。

## 3.3 核心概念 ##

**① Executor与Core**

`Spark Executor`是集群中运行在工作节点（`Worker`）中的一个`JVM`进程，是整个集群中的专门用于计算的节点。在提交应用中，可以提供参数指定计算节点的个数，以及对应的资源。这里的资源一般指的是工作节点`Executor`的内存大小和使用的虚拟`CPU`核（`Core`）数量。

应用程序相关启动参数如下：

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>–num-executors</td> 
   <td>配置Executor的数量</td> 
  </tr> 
  <tr> 
   <td>–executor-memory</td> 
   <td>配置每个Executor的内存大小</td> 
  </tr> 
  <tr> 
   <td>–executor-cores</td> 
   <td>配置每个Executor的虚拟CPU core数量</td> 
  </tr> 
 </tbody> 
</table>

**② 并行度（Parallelism）**

在分布式计算框架中一般都是多个任务同时执行，由于任务分布在不同的计算节点进行计算，所以能够真正地实现多任务并行执行，记住，这里是并行，而不是并发。这里我们将整个集群并行执行任务的数量称之为并行度。那么一个作业到底并行度是多少呢？这个取决于框架的默认配置。应用程序也可以在运行过程中动态修改。

**③ 有向无环图（DAG）**  
![在这里插入图片描述][20200922164144120.png_pic_center]  
`DAG`（`Directed Acyclic Graph`）有向无环图是由点和线组成的拓扑图形，该图形具有方向，不会闭环。

`Spark`计算引擎的特点主要是 `Job`内部的`DAG`支持（不跨越`Job`），以及实时计算。这里所谓的有向无环图，并不是真正意义的图形，而是由`Spark`程序直接映射成的数据流的高级抽象模型。简单理解就是将整个程序计算的执行过程用图形表示出来,这样更直观，更便于理解，可以用于表示程序的拓扑结构。

## 3.4 提交流程 ##

所谓的提交流程，其实就是我们开发人员根据需求写的应用程序通过`Spark`客户端提交给`Spark`运行环境执行计算的流程。在不同的部署环境中，这个提交过程基本相同，但是又有细微的区别，我们这里不进行详细的比较，但是因为国内工作中，将`Spark`引用部署到`Yarn`环境中会更多一些，所以本课程中的提交流程是基于`Yarn`环境的。  
![在这里插入图片描述][2020092216231495.png]  
`Spark`应用程序提交到`Yarn`环境中执行的时候，一般会有两种部署执行的方式：`Client`和`Cluster`。两种模式主要区别在于：`Driver`程序的运行节点。

**① Yarn Client模式**

`Client`模式将用于监控和调度的`Driver`模块在客户端执行，而不是`Yarn`中，所以一般用于测试。

1.  `Driver`在任务提交的本地机器上运行
2.  `Driver`启动后会和`ResourceManager`通讯申请启动`ApplicationMaster`
3.  `ResourceManager`分配`container`，在合适的`NodeManager`上启动`ApplicationMaster`，负责向`ResourceManager`申请`Executor`内存
4.  `ResourceManager`接到`ApplicationMaster`的资源申请后会分配`container`，然后`ApplicationMaster`在资源分配指定的`NodeManager`上启动`Executor`进程
5.  `Executor`进程启动后会向`Driver`反向注册，`Executor`全部注册完成后`Driver`开始执行`main`函数
6.  之后执行到`Action`算子时，触发一个`Job`，并根据宽依赖开始划分`stage`，每个`stage`生成对应的`TaskSet`，之后将`task`分发到各个`Executor`上执行

**② Yarn Cluster模式**

`Cluster`模式将用于监控和调度的`Driver`模块启动在`Yarn`集群资源中执行。一般应用于实际生产环境。

1.  在`YARN Cluster`模式下，任务提交后会和`ResourceManager`通讯申请启动`ApplicationMaster`
2.  随后`ResourceManager`分配`container`，在合适的`NodeManager`上启动`ApplicationMaster`，此时的`ApplicationMaster`就是`Driver`
3.  `Driver`启动后向`ResourceManager`申请`Executor`内存，`ResourceManager`接到`ApplicationMaster`的资源申请后会分配`container`，然后在合适的`NodeManager`上启动`Executor`进程
4.  `Executor`进程启动后会向`Driver`反向注册，`Executor`全部注册完成后`Driver`开始执行`main`函数
5.  之后执行到`Action`算子时，触发一个`Job`，并根据宽依赖开始划分`stage`，每个`stage`生成对应的`TaskSet`，之后将`task`分发到各个`Executor`上执行

[20200917152353231.png]: /images/20221123/a9fa305ef298467a8335cd9df7299f98.png
[20200917145844473.png]: /images/20221123/b0d1ca476b704fd4aebef83e763ae38f.png
[2020092215213139.png]: /images/20221123/5b7bbeb1350c476d8aaf878e4d3a1cc2.png
[20200922160459721.png]: /images/20221123/969e0f2554c84d4bbd928925a979c152.png
[20200922155530323.png]: /images/20221123/bbaa4710b0bc4023b6e70459efdfc42a.png
[Link 1]: https://spark.apache.org/docs/latest/running-on-kubernetes.html
[20200922155631730.png]: /images/20221123/65f9913de037405fb91a485a1e46a08a.png
[20200922150832259.png]: /images/20221123/67330b87745a4307853d4665a367772a.png
[20200922160306871.png]: /images/20221123/2dd0fe99f10e464598aa6e18ef81fc89.png
[20200922164144120.png_pic_center]: /images/20221123/d97109ca87054141ae2d6b0eb32fd74f.png
[2020092216231495.png]: /images/20221123/562b5a84f76e4330bc23889d1b070b9d.png