浅谈Spark中Job-Stage-Task之间的关系

浅浅的花香味﹌ 2023-01-19 14:50 184阅读 0赞

### 目录 ###

*   *  （1）Spark-Job-Stage-Task之间的关系
     *  （2）RDD-Partition-Records-Task之间的关系

## （1）Spark-Job-Stage-Task之间的关系 ##

在开始之前需要先了解Spark中Application，Job，Stage等基本概念，官方给出的解释如下表：

<table> 
 <thead> 
  <tr> 
   <th>Term</th> 
   <th>Meaning</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Application</td> 
   <td>用户编写的Spark应用程序,包括一个Driver和多个executors</td> 
  </tr> 
  <tr> 
   <td>Driver Program</td> 
   <td>运行main()函数并创建SparkContext进程</td> 
  </tr> 
  <tr> 
   <td>Cluster manager</td> 
   <td>在集群上获取资源的外部服务，如standalone ,yarn,Mesos</td> 
  </tr> 
  <tr> 
   <td>worker node</td> 
   <td>集群中可以运行程序代码的节点（机器）</td> 
  </tr> 
  <tr> 
   <td>Executor</td> 
   <td>运行在worker node上执行具体的计算任务，存储数据的进程</td> 
  </tr> 
  <tr> 
   <td>Task</td> 
   <td>被分配到一个Executor上的计算单元</td> 
  </tr> 
  <tr> 
   <td>Job</td> 
   <td>由多个任务组成的并行计算阶段，因RDD的Action产生</td> 
  </tr> 
  <tr> 
   <td>Stage</td> 
   <td>每个Job被分为小的计算任务组，每组称为一个stage</td> 
  </tr> 
  <tr> 
   <td>DAGScheduler</td> 
   <td>根据Job构建基于Stage的DAG，并提交Stage给TaskScheduler</td> 
  </tr> 
  <tr> 
   <td>TaskScheduler</td> 
   <td>将TaskSet提交给worker运行，每个executor运行什么task在此分配</td> 
  </tr> 
 </tbody> 
</table>

如下图所示，**一个Spark程序可以被划分为一个或多个Job，划分的依据是RDD的Action算子，每遇到一个RDD的Action操作就生成一个新的Job。**

每个spark Job在具体执行过程中因为shuffle的存在，需要将其划分为一个或多个可以并行计算的stage，划分的依据是RDD间的依赖关系，当遇到宽依赖（Wide Dependency）时因需要进行shuffle操作，这涉及到了不同Partition之间进行数据合并，故以此为界划分不同的Stage。Stage是由**Task组**组成的并行计算，因此每个stage中可能存在多个Task，这些Task执行相同的程序逻辑，只是它们操作的数据不同。一般RDD的一个Partition对应一个Task,**Task可以分为ResultTask和ShuffleMapTask。**

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

## （2）RDD-Partition-Records-Task之间的关系 ##

**通常一个RDD被划分为一个或多个Partition，Partition是Spark进行数据处理的基本单位，一般来说一个Partition对应一个Task，而一个Partition中通常包含数据集中的多条记录(Record)。** 注意不同Partition中包含的记录数可能不同。Partition的数目可以在创建RDD时指定，也可以通过reparation和coalesce等算子重新进行划分。

通常在进行shuffle的时候也会重新进行分区，这是对于key-value RDD，Spark通常根据RDD中的Partitioner来进行分区，目前Spark中实现的Partitioner有两种：**HashPartitioner和RangePartitioner**，当然也可以实现自定义的Partitioner，只需要继承抽象类Partitioner并实现`numPartitions and getPartition(key: Any)`即可。

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

将上面的概念串联起来，可以得到下面的运行层次图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 2]

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

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

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70]: /images/20221021/7d29305fda6c4738a00f83220ffe93ef.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 1]: /images/20221021/79a64a39ba2c45d2a1743d02fb41af5c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70 2]: /images/20221021/6881ae1880f34af4a92b552f0982c707.png