Spark工作原理及基础概念（超详细！）

墨蓝 2022-12-23 07:21 137阅读 0赞

### 目录 ###

*   *  一、Spark概述
     *   *  （1）概述
         *  （2）Spark整体架构
         *  （3）Spark特性
         *  （4）Spark与MR
         *  （5）Spark Streaming与Storm
         *  （6）Spark SQL与Hive
     *  二、Spark基本原理
     *   *  （1）Spark Core
         *  （2）Spark SQL
         *  （3）Spark Streaming
         *  （4）Spark基本工作原理
         *   *  （1）分布式数据集
             *  （2）弹性
             *  （3）迭代式处理
             *  （4）容错性
     *  三、Spark 运行模式及集群角色
     *   *  （1）Spark运行模式
         *  （2）Spark集群角色
         *   *  （1）Cluster Manager
             *  （2）worker
             *  （3）executor
             *  （4）Application
             *  （5）Driver
             *  （6）sparkContext
         *  （3）Spark其他核心概念
         *   *  （1）RDD
             *  （2）DAG
             *  （3）DAG Scheduler
             *  （4）Task Scheduler
             *  （5）Job
             *  （6）stage
             *  （7）TaskSet Task
     *  四、Spark作业运行流程
     *   *  （1）Spark作业运行流程
         *  （2） Spark RDD迭代过程
         *  （3）Yarn资源管理器介绍
         *  （4）Spark程序在Yarn上执行流程
     *  五、基于WordCount程序开发
     *   *  （1）Java开发WordCount程序
         *  （2）Scala开发WordCount程序
         *  （3）程序打包提交Spark集群测试

## 一、Spark概述 ##

### （1）概述 ###

Spark，是一种"One Stack to rule them all"的大数据计算框架，期望使用一个技术堆栈就完美地解决大数据领域的各种计算任务。Apache官方，对Spark的定义就是：通用的大数据快速处理引擎。  
Spark使用Spark RDD、Spark SQL、 Spark Streaming，MLlib，GraphX成功解决了大数据领城中，离线批处理、交互式查询、实时流计算、机器学习与图计算等最重要的任务和问题。  
Spark除了一站式的特点之外，另外一个最重要的特点，就是基于内存进行计算，从而让它的速度可以达到MapReduce、Hive的数倍甚至数十倍！  
现在已经有很多大公司正在生产环境下深度地使用Spark作为大数据的计算框架，包括eBay.Yahool、 BAT、网易、京东、华为、大众点评、优酷土豆、搜狗等等。  
Spark同时也获得了多个世界顶级IT厂商的支持，包括IBM、 Intel等。

Spark，是一种通用的大数据计算框架，I正如传统大数据技术Hadoop的MapReduce、Hive引擎，以及Storm流式实时计算引擎等，  
Spark包含了大数据领城常见的各种计算框架：比如Spark Core用于离线计算，Spark SQL用于交互式查询，Spark Streaming用于实时流式计算，Spark MILlib用于机器学习，Spark GraphX用于图计算。  
Spark主要用于大数据的计算，而Hadoop以后主要用于大数据的存储（比如HDFS、Hive，HBase等），以及资源调度（Yarn）。  
Spark+Hadoop的组合，是未来大数据领域最热门的组合，也是最有前景的组合！

### （2）Spark整体架构 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]

### （3）Spark特性 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 1]

**（1）spark 计算速度快**

spark将每个任务构建成DAG进行计算，内部的计算过程通过弹性式分布式数据集RDD在内存在进行计算，相比于hadoop的mapreduce效率提升了100倍。

**（2）易于使用**

spark 提供了大量的算子，开发只需调用相关api进行实现无法关注底层的实现原理。

通用的大数据解决方案

相较于以前离线任务采用mapreduce实现，实时任务采用storm实现，目前这些都可以通过spark来实现，降低来开发的成本。同时spark 通过spark SQL降低了用户的学习使用门槛，还提供了机器学习，图计算引擎等。

**（3）支持多种的资源管理模式**

学习使用中可以采用local 模型进行任务的调试，在正式环境中又提供了standalone，yarn等模式，方便用户选择合适的资源管理模式进行适配。

**（4）社区支持**

spark 生态圈丰富，迭代更新快，成为大数据领域必备的计算引擎。

### （4）Spark与MR ###

MapReduce能够完成的各种离线批处理功能，以及常见算法（比如二次排序、topn等），基于Spark RDD的核心编程，都可以实现，并且可以更好地、更容易地实现。而且基于Spark RDD编写的离线批处理程序，运行速度是MapReduce的数倍，速度上有非常明显的优势。  
Spark相较于MapReduce速度快的最主要原因就在于，MapReduce的计算模型太死板，必须是map-reduce模式，有时候即使完成一些诸如过滤之类的操作，也必须经过map-reduce过程，这样就必须经过shuffle过程。而  
MapReduce的shuffle过程是最消耗性能的，因为shuffle中间的过程必须基于磁盘来读写。而Spark的shuffle虽然也要基于磁盘，但是其大量transformation操作，比如单纯的map或者filter等操作，可以直接基于内存进行pipeline操作，速度性能自然大大提升。  
但是Spark也有其劣势。由于Spark基于内存进行计算，虽然开发容易，但是真正面对大数据的时候（比如一次操作针对10亿以上级别），在没有进行调优的情况下，可能会出现各种各样的问题，比如OOM内存溢出等等。导致Spark程序可能都无法完全运行起来，就报错挂掉了，而MapReduce即使是运行缓慢，但是至少可以慢慢运行完。  
此外，Spark由于是新崛起的技术新秀，因此在大数据领域的完善程度，肯定不如MapReduce，比如基于HBase、Hive作为离线批处理程序的输入输出，Spark就远没有MapReduce来的完善。实现起来非常麻烦。

### （5）Spark Streaming与Storm ###

Spark Streaming 与Storm都可以用于进行实时流计算。但是他们两者的区别是非常大的。其中区别之一，就是，Spark  
Streaming 和Storm的计算模型完全不一样，Spark Streaming是基于RDD的，因此需要将一小段时间内的，比如1秒内的数据，收集起来，作为一个RDD，然后再针对这个batch的数据进行处理。而Storm却可以做到每来一条数据，都可以立即进行处理和计算。因此，Spark Streaming实际上严格意义上来说，只能称作准实时的流计算框架；西Storm是真正意义上的实时计算框架  
此外，Storm支持的一项高级特性，是Spark Streaming暂时不具备的，即Storm支持在分布式流式计算程序（Topolopy）在运行过程中，可以动态地调整并行度，从而动态提高并发处理能力。而Spark Streaming是无法动态调整并行度的。  
但是Spark Streaming也有其优点，首先Spark Streaming由于是基于batch进行处理的，因此相较于Storm基于单条数据进行处理，具有数倍甚至数十倍的吞吐量。  
此外，Spark Streaming由于也身处于Spark生态圈内，因此Spark Streaming可以与Spark Core、 Spark SQL，甚至是Spark Mllib.Spark GraphX进行无缝整合。流式处理完的数据，可以立即进行各种map、reduce转换操作，可以立即使用sql进行查询，甚至可以立即使用machine learning或者图计算算法进行处理。这种一站式的大数据处理功能和优势，是Storm无法匹敌的。  
因此，综合上述来看，通常在对实时性要求特别高，而且实时数据量不稳定，比如在白天有高峰期的情况下，可以选择使用Storm。但是如果是对实时性要求一般，允许1秒的准实时处理，而且不要求动态调整并行度的话，选择Spark Streaming是更好的选择。

### （6）Spark SQL与Hive ###

Spark SQL实际上并不能完全替代Hive,因为Hive是一种基于HDFS的数据仓库,并且提供了基于SQL模型的,针对存储了大数据的数据仓库,进行分布式交互查询的查询引擎。  
严格的来说, Spark SQL能够替代的,是ive的查询引擎,而不是Hive本身,实际上即使在生产环境下, SparkSQL也是针对Hive数据仓库中的数据进行查询, Spark本身自己是不提供存储的,自然也不可能替代Hive作为数据仓库的这个功能。  
Spark SQL的一个优点,相较于Hive查询引擎来说,就是速度快,同样的SQL语句,可能使用Hive的查询引擎,由于其底层基于 MapReduce,必须经过 shuffle过程走磁盘,因此速度是非常缓慢的。很多复杂的SQL语句,在hive中执行都需要一个小时以上的时间。而 Spark SQLSpark由于其底层基于自身的基于内存的特点,因此速度达到了Hive查询引擎的数倍以上。  
而 Spark SQL相较于Hive的另外一个优点,就是支持大量不同的数据源,包括ive、json、 parquet、jdbc等等此外, Spark SQLSpark由于身处技术堆栈内,也是基于RDD来工作,因此可以与 Spark的其他组件无缝整合使用,配合起来实现许多复杂的功能。比如 Spark SQL支持可以直接针对hdfs文件执行sq语句!

## 二、Spark基本原理 ##

### （1）Spark Core ###

Spark Core是Spark的核心，其包含如下几个部分：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 2]  
**（1）spark 基础配置**

sparkContext是spark应用程序的入口，spark应用程序的提交和执行离不开sparkContext，它隐藏了网络通信，分布式部署，消息通信，存储体系，计算存储等，开发人员只需要通过sparkContext等api进行开发即可。

sparkRpc 基于netty实现，分为异步和同步两种方式。事件总线主要用于sparkContext组件间的交换，它属于监听者模式，采用异步调用。度量系统主要用于系统的运行监控。

**（2）spark 存储系统**

它用于管理spark运行中依赖的数据存储方式和存储位置，spark的存储系统优先考虑在各节点以内存的方式存储数据，内存不足时将数据写入磁盘中，这也是spark计算性能高的重要原因。

我们可以灵活的控制数据存储在内存还是磁盘中，同时可以通过远程网络调用将结果输出到远程存储中，比如hdfs，hbase等。

**（3）spark 调度系统**

spark 调度系统主要由DAGScheduler和TaskScheduler组成。

DAGScheduler 主要是把一个Job根据RDD间的依赖关系，划分为多个Stage，对于划分后的每个Stage都抽象为一个或多个Task组成的任务集，并交给TaskScheduler来进行进一步的任务调度。而TaskScheduler 负责对每个具体的Task进行调度。

具体调度算法有FIFO，FAIR：

*  FIFO调度：先进先出，这是Spark默认的调度模式。
 *  FAIR调度：支持将作业分组到池中，并为每个池设置不同的调度权重，任务可以按照权重来决定执行顺序。

### （2）Spark SQL ###

spark sql提供了基于sql的数据处理方法，使得分布式的数据集处理变的更加简单，这也是spark 广泛使用的重要原因。

目前大数据相关计算引擎一个重要的评价指标就是：是否支持sql，这样才会降低使用者的门槛。spark sql提供了两种抽象的数据集合DataFrame和DataSet。

DataFrame 是spark Sql 对结构化数据的抽象，可以简单的理解为spark中的表，相比较于RDD多了数据的表结构信息(schema).DataFrame = Data + schema

RDD是分布式对象集合，DataFrame是分布式Row的集合，提供了比RDD更丰富的算子，同时提升了数据的执行效率。

DataSet 是数据的分布式集合 ，它具有RDD强类型的优点 和Spark SQL优化后执行的优点。DataSet可以由jvm对象构建，然后使用map，filter，flatmap等操作函数操作。

关于Spark SQL可以看这篇文章：  
[https://blog.csdn.net/weixin\_45366499/article/details/108749586][https_blog.csdn.net_weixin_45366499_article_details_108749586]

### （3）Spark Streaming ###

这个模块主要是对流数据的处理，支持流数据的可伸缩和容错处理，可以与Flume和Kafka等已建立的数据源集成。Spark Streaming的实现，也使用RDD抽象的概念，使得在为流数据编写应用程序时更为方便。

关于Spark Streaming可以看这篇文章：  
[https://blog.csdn.net/weixin\_45366499/article/details/108816335][https_blog.csdn.net_weixin_45366499_article_details_108816335]

### （4）Spark基本工作原理 ###

Spark基本工作原理的理解，其最主要的是要搞清楚什么是RDD以及RDD的特性。深刻理解了RDD的特性，也就理解了数据在spark中是如何被处理的（spark的基本工作原理）

那么RDD是什么，官方说法：  
RDD是Spark提供的核心抽象，全称为Resillient Distributed Dataset，即弹性分布式数据集。

最简单的理解：  
RDD就是源数据的抽象，或者叫映射，或者就代表。也就是说，数据要被spark进行处理，在处理之前的首要任务就是要将数据映射成RDD，对于spark来说，RDD才是我们处理数据的规则，我只认RDD，只有RDD，通过我spark的计算引擎，才能发挥巨大的威力！

#### （1）分布式数据集 ####

RDD是Spark提供的核心抽象，全称为Resillient Distributed Dataset，即弹性分布式数据集。

RDD在抽象上来说是一种元素集合，包含了数据。它是被分区的，分为多个分区，每个分区分布在集群中的不同节点上，从而让RDD中的数据可以被并行操作。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 3]

#### （2）弹性 ####

RDD的数据默认情况下存放在内存中的，但是在内存资源不足时，Spark会自动将RDD数据写入磁盘。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 4]

#### （3）迭代式处理 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 5]  
对节点1、2、3、4上的数据进行处理完成之后，可能会移动到其他的节点内存中继续处理！Spark 与Mr最大的不同在与迭代式计算模型：Mr分为两个阶段，map和reduce，两个阶段处理完了就结束了，所以我们在一个job中能做的处理很有限，只能在map和reduce中处理；而spark计算过程可以分为n个阶段，因为他是内存迭代式的，我们在处理完一个阶段之后，可以继续往下处理很多阶段，而不是两个阶段。所以Spark相较于MR，计算模型可以提供更强大的功能。

#### （4）容错性 ####

RDD最重要的特性就是，提供了容错性，可以自动从节点失败中恢复过来。即如果某个节点上的RDD partition，因为节点故障，导致数据丢了，那么RDD会自动通过自己的数据来源重新计算该partition。这一切对使用者是透明的。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 6]

## 三、Spark 运行模式及集群角色 ##

### （1）Spark运行模式 ###

<table> 
 <thead> 
  <tr> 
   <th>运行模式</th> 
   <th>运行类型</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>local</td> 
   <td>本地模式</td> 
   <td>常用于本地开发测试，分为local单线程和local-cluster多线程模式</td> 
  </tr> 
  <tr> 
   <td>standalone</td> 
   <td>集群模式</td> 
   <td>独立模式，在spark自己的资源调度管理框架上运行，该框架采用master/salve结构</td> 
  </tr> 
  <tr> 
   <td>yarn</td> 
   <td>集群模式</td> 
   <td>在yarn资源管理器框架上运行，由yarn负责资源管理，spark负责任务调度和计算</td> 
  </tr> 
  <tr> 
   <td>mesos</td> 
   <td>集群模式</td> 
   <td>在mesos资源管理器框架上运行，由mesos负责资源管理，spark负责任务调度和计算</td> 
  </tr> 
  <tr> 
   <td>k8s</td> 
   <td>集群模式</td> 
   <td>在k8s上运行</td> 
  </tr> 
 </tbody> 
</table>

### （2）Spark集群角色 ###

下图是spark的集群角色图，主要有集群管理节点cluster manager，工作节点worker，执行器executor，驱动器driver和应用程序application 五部分组成，下面详细说明每部分的特点。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 7]

#### （1）Cluster Manager ####

集群管理器，它存在于Master进程中，主要用来对应用程序申请的资源进行管理，根据其部署模式的不同，可以分为local，standalone，yarn，mesos等模式。

#### （2）worker ####

worker是spark的工作节点，用于执行任务的提交，主要工作职责有下面四点：

*  worker节点通过注册机向cluster manager汇报自身的cpu，内存等信息。
 *  worker 节点在spark master作用下创建并启用executor，executor是真正的计算单元。
 *  spark master将任务Task分配给worker节点上的executor并执行运用。
 *  worker节点同步资源信息和executor状态信息给cluster manager。

#### （3）executor ####

executor 是真正执行计算任务的组件，它是application运行在worker上的一个进程。这个进程负责Task的运行，它能够将数据保存在内存或磁盘存储中，也能够将结果数据返回给Driver。

#### （4）Application ####

application是Spark API 编程的应用程序，它包括实现Driver功能的代码和在程序中各个executor上要执行的代码，一个application由多个job组成。其中应用程序的入口为用户所定义的main方法。

#### （5）Driver ####

驱动器节点，它是一个运行Application中main函数并创建SparkContext的进程。application通过Driver 和Cluster Manager及executor进行通讯。它可以运行在application节点上，也可以由application提交给Cluster Manager，再由Cluster Manager安排worker进行运行。

Driver节点也负责提交Job，并将Job转化为Task，在各个Executor进程间协调Task的调度。

#### （6）sparkContext ####

sparkContext是整个spark应用程序最关键的一个对象，是Spark所有功能的主要入口点。核心作用是初始化spark应用程序所需要的组件，同时还负责向master程序进行注册等。

### （3）Spark其他核心概念 ###

#### （1）RDD ####

它是Spark中最重要的一个概念，是弹性分布式数据集，是一种容错的、可以被并行操作的元素集合，是Spark对所有数据处理的一种基本抽象。可以通过一系列的算子对rdd进行操作，主要分为Transformation和Action两种操作。

*  ‍‍‍‍‍Transformation(转换)：是对已有的RDD进行换行生成新的RDD，对于转换过程采用惰性计算机制，不会立即计算出结果。常用的方法有map，filter，flatmap等。
 *  Action(执行)：对已有对RDD对数据执行计算产生结果，并将结果返回Driver或者写入到外部存储中。常用到方法有reduce，collect，saveAsTextFile等。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 9]

#### （2）DAG ####

DAG是一个有向无环图，在Spark中， 使用 DAG 来描述我们的计算逻辑。主要分为DAG Scheduler 和Task Scheduler。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 10]

#### （3）DAG Scheduler ####

DAG Scheduler 是面向stage的高层级的调度器，DAG Scheduler把DAG拆分为多个Task，每组Task都是一个stage，解析时是以shuffle为边界进行反向构建的，每当遇见一个shuffle，spark就会产生一个新的stage，接着以TaskSet的形式提交给底层的调度器（task scheduler），每个stage封装成一个TaskSet。DAG Scheduler需要记录RDD被存入磁盘物化等动作，同时会需要Task寻找最优等调度逻辑，以及监控因shuffle跨节点输出导致的失败。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 11]

#### （4）Task Scheduler ####

Task Scheduler 负责每一个具体任务的执行。它的主要职责包括

*  任务集的调度管理
 *  状态结果跟踪
 *  物理资源调度管理
 *  任务执行
 *  获取结果

#### （5）Job ####

job是有多个stage构建的并行的计算任务，job是由spark的action操作来触发的，在spark中一个job包含多个RDD以及作用在RDD的各种操作算子。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 12]

#### （6）stage ####

DAG Scheduler会把DAG切割成多个相互依赖的Stage，划分Stage的一个依据是RDD间的宽窄依赖。

在对Job中的所有操作划分Stage时，一般会按照倒序进行，即从Action开始，遇到窄依赖操作，则划分到同一个执行阶段，遇到宽依赖操作，则划分一个新的执行阶段，且新的阶段为之前阶段的parent，然后依次类推递归执行。

child Stage需要等待所有的parent Stage执行完之后才可以执行，这时Stage之间根据依赖关系构成了一个大粒度的DAG。在一个Stage内，所有的操作以串行的Pipeline的方式，由一组Task完成计算。

#### （7）TaskSet Task ####

TaskSet 可以理解为一种任务，对应一个stage，是Task组成的任务集。一个TaskSet中的所有Task没有shuffle依赖可以并行计算。

Task是spark中最独立的计算单元，由Driver Manager发送到executer执行，通常情况一个task处理spark RDD一个partition。Task分为ShuffleMapTask和ResultTask两种，位于最后一个Stage的Task为ResultTask，其他阶段的属于ShuffleMapTask。

## 四、Spark作业运行流程 ##

### （1）Spark作业运行流程 ###

spark应用程序以进程集合为单位在分布式集群上运行，通过driver程序的main方法创建sparkContext的对象与集群进行交互。具体运行流程如下：

*  sparkContext向cluster Manager申请CPU，内存等计算资源。
 *  cluster Manager分配应用程序执行所需要的资源，在worker节点创建executor。
 *  sparkContext将程序代码和task任务发送到executor上进行执行，代码可以是编译成的jar包或者python文件等。接着sparkContext会收集结果到Driver端。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 13]

### （2） Spark RDD迭代过程 ###

*  sparkContext创建RDD对象，计算RDD间的依赖关系，并组成一个DAG有向无环图。
 *  DAGScheduler将DAG划分为多个stage，并将stage对应的TaskSet提交到集群的管理中心，stage的划分依据是RDD中的宽窄依赖，spark遇见宽依赖就会划分为一个stage，每个stage中包含来一个或多个task任务，避免多个stage之间消息传递产生的系统开销。
 *  taskScheduler 通过集群管理中心为每一个task申请资源并将task提交到worker的节点上进行执行。
 *  worker上的executor执行具体的任务。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 14]

### （3）Yarn资源管理器介绍 ###

spark 程序一般是运行在集群上的，spark on yarn是工作或生产上用的非常多的一种运行模式。

没有yarn模式前，每个分布式框架都要跑在一个集群上面，比如说Hadoop要跑在一个集群上，Spark用集群的时候跑在standalone上。这样的话整个集群的资源的利用率低，且管理起来比较麻烦。

yarn是分布式资源管理和任务管理管理，主要由ResourceManager，NodeManager和ApplicationMaster三个模块组成。

NodeManager 负责节点的维护，执行和监控task运行状况。会通过心跳的方式向ResourceManager汇报自己的资源使用情况。

yarn资源管理器的每个节点都运行着一个NodeManager，是ResourceManager的代理。如果主节点的ResourceManager宕机后，会连接ResourceManager的备用节点。

ApplicationMaster 负责具体应用程序的调度和资源的协调，它会与ResourceManager协商进行资源申请。ResourceManager以container容器的形式将资源分配给application进行运行。同时负责任务的启停。

container 是资源的抽象，它封装着每个节点上的资源信息(cpu,内存，磁盘，网络等)，yarn将任务分配到container上运行，同时该任务只能使用container描述的资源，达到各个任务间资源的隔离。

### （4）Spark程序在Yarn上执行流程 ###

spark on yarn分为两种模式yarn-client模式，和yarn—cluster模式，一般线上采用的是yarn-cluster模式。

**（1）yarn-client模式**  
driver在客户端本地执行，这种模式可以使得spark application和客户端进行交互，因为driver在客户端可以通过webUI访问driver的状态。同时Driver会与yarn集群中的Executor进行大量的通信，会造成客户机网卡流量的大量增加。

**（2）yarn-cluster模式**  
Yarn-Cluster主要用于生产环境中，因为Driver运行在Yarn集群中某一台NodeManager中，每次提交任务的Driver所在的机器都是随机的，不会产生某一台机器网卡流量激增的现象，缺点是任务提交后不能看到日志。只能通过yarn查看日志。

下图是yarn-cluster运行模式：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 16]

*  client 向yarn提交应用程序，包含ApplicationMaster程序、启动ApplicationMaster的命令、需要在Executor中运行的程序等。
 *  ApplicationMaster程序启动ApplicationMaster的命令、需要在Executor中运行的程序等。
 *  ApplicationMaster向ResourceManager注册，这样用户可以直接通过ResourceManage查看应用程序的运行状态。
 *  ApplicationMaster申请到资源（也就是Container）后，便与对应的NodeManager通信，启动Task。
 *  Task向ApplicationMaster汇报运行的状态和进度，以让ApplicationMaster随时掌握各个任务的运行状态，从而可以在任务失败时重新启动任务。
 *  应用程序运行完成后，ApplicationMaster向ResourceManager申请注销并关闭自己。

## 五、基于WordCount程序开发 ##

下面使用Java和Scala语言编写WordCount程序  
关于spark idea环境配置或安装请看这篇文章：  
[https://blog.csdn.net/weixin\_45366499/article/details/107957186][https_blog.csdn.net_weixin_45366499_article_details_107957186]  
[https://blog.csdn.net/weixin\_45366499/article/details/108518504][https_blog.csdn.net_weixin_45366499_article_details_108518504]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 17]

### （1）Java开发WordCount程序 ###

package com.kfk.spark;
    
    import org.apache.spark.SparkConf;
    import org.apache.spark.api.java.JavaPairRDD;
    import org.apache.spark.api.java.JavaRDD;
    import org.apache.spark.api.java.JavaSparkContext;
    import org.apache.spark.api.java.function.FlatMapFunction;
    import org.apache.spark.api.java.function.Function2;
    import org.apache.spark.api.java.function.PairFunction;
    import org.apache.spark.api.java.function.VoidFunction;
    import scala.Tuple2;
    
    import java.util.Arrays;
    import java.util.Iterator;
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/11/22 * @time : 10:19 下午 */
    public class WordCountJava { 
        public static void main(String[] args) { 
    
            SparkConf sparkConf = new SparkConf().setAppName("wordCountApp").setMaster("local");
    
            JavaSparkContext sc = new JavaSparkContext(sparkConf);
    
            JavaRDD lines = sc.textFile("hdfs://bigdata-pro-m04:9000/user/caizhengjie/datas/wordcount.txt");
    
            JavaRDD words = lines.flatMap(new FlatMapFunction<String,String>() { 
                public Iterator call(String line) throws Exception { 
                    return Arrays.asList(line.split(" ")).iterator();
                }
            });
    
            JavaPairRDD word = words.mapToPair(new PairFunction<String,String,Integer>() { 
                public Tuple2 call(String word) throws Exception { 
                    return new Tuple2(word,1);
                }
            });
    
            JavaPairRDD wordcount = word.reduceByKey(new Function2<Integer,Integer,Integer>() { 
                public Integer call(Integer v1, Integer v2) throws Exception { 
                    return v1+ v2;
                }
            });
    
            wordcount.foreach(new VoidFunction<Tuple2<String,Integer>>() { 
                public void call(Tuple2<String,Integer> o) throws Exception { 
                    System.out.println(o._1 + " : " + o._2);
                }
            });
    
        }
    }

运行结果：

spark : 1
    hive : 3
    hadoop : 3
    python : 1
    flink : 2
    java : 5
    storm : 1
    hbase : 1

### （2）Scala开发WordCount程序 ###

package com.kfk.spark
    
    import org.apache.spark.{ SparkConf, SparkContext}
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2020/11/22 * @time : 10:48 下午 */
    object WordCountScala { 
        def main(args: Array[String]): Unit = { 
            val sparkConf = new SparkConf().setAppName("wordCountApp").setMaster("local")
            val sc = new SparkContext(sparkConf)
            val lines = sc.textFile("hdfs://bigdata-pro-m04:9000/user/caizhengjie/datas/wordcount.txt")
    
            // val wordcount = lines.flatMap(line => line.split(" ")).map(word => (word,1)).reduceByKey((x,y) => (x+y)).foreach((_wordcount => println(_wordcount._1 + " : " + _wordcount._2))
    
            val words = lines.flatMap(line => line.split(" "))
            val word = words.map(word => (word,1))
            val wordcount = word.reduceByKey((x,y) => (x+y))
    
            // println(wordcount.collect().toList)
    
            wordcount.foreach(_wordcount => println(_wordcount._1 + " : " + _wordcount._2))
        }
    }

运行结果：

spark : 1
    hive : 3
    hadoop : 3
    python : 1
    flink : 2
    java : 5
    storm : 1
    hbase : 1

### （3）程序打包提交Spark集群测试 ###

bin/spark-submit --class com.kfk.spark.WordCountJava --master local[2] /opt/jars/wordcountjava.jar

运行结果：

spark : 1
    hive : 3
    hadoop : 3
    python : 1
    flink : 2
    java : 5
    storm : 1
    hbase : 1

参考文章：  
[https://blog.csdn.net/QcloudCommunity/article/details/109685416][https_blog.csdn.net_QcloudCommunity_article_details_109685416]

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/59861453c49a478a8dd2b4799254ba4c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/883b6ece5d74484f9d6607abf342bd03.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221120/2f63439e584648aab11f7afe35583e7c.png
[https_blog.csdn.net_weixin_45366499_article_details_108749586]: https://blog.csdn.net/weixin_45366499/article/details/108749586
[https_blog.csdn.net_weixin_45366499_article_details_108816335]: https://blog.csdn.net/weixin_45366499/article/details/108816335
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221120/1076abd43ada4089b407aa82a296ae2d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221120/9c9ec4e3e1c4494db18ab177b32a41c9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221120/988c2f2f77e04edf98af41740ea1b352.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221120/1346b204b90b4c749eb79a782de64a5f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 7]: /images/20221120/d4ef32f35d954c158de6ecfe0c624de4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 8]: /images/20221120/76216f4c93654c4d8f95677733f9ac0f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 9]: /images/20221120/eed3b893a3da4cf5a20ce7acec0f83cf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 10]: /images/20221120/28de658f4df34d09b99e40ed0e81c3cd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 11]: /images/20221120/dd8af248801f4c4285d7296dd0805cf1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 12]: /images/20221120/afc95d3a4a45418698b54471716315b4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 13]: /images/20221120/f305036a0fd14ba687a758ab5f315136.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 14]: /images/20221120/3bfbe176306c47d8b981e712dd248a2d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 15]: /images/20221120/b457a385f8944376a6f959cb33746394.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 16]: /images/20221120/d7d3d16cea5645e6a51d135cf3a314c6.png
[https_blog.csdn.net_weixin_45366499_article_details_107957186]: https://blog.csdn.net/weixin_45366499/article/details/107957186
[https_blog.csdn.net_weixin_45366499_article_details_108518504]: https://blog.csdn.net/weixin_45366499/article/details/108518504
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 17]: /images/20221120/f5ce1ad3f210403cae0ff1ffb50a7c4d.png
[https_blog.csdn.net_QcloudCommunity_article_details_109685416]: https://blog.csdn.net/QcloudCommunity/article/details/109685416