spark内核篇-task数与并行度

每一个 spark job 根据 shuffle 划分 stage，每个 stage 形成一个或者多个 taskSet，了解了每个 stage 需要运行多少个 task，有助于我们优化 spark 运行

task 数 

首先需要了解以下概念：

RDD，弹性分布式数据集，多个 partition；

split，切片，HDFS 上文件为什么要切片，如何切片，参考我的博客 hadoop 的 Split；

textFlie 分区，textFile 如何对一个文件分区，参考我的博客 RDD认知与创建；

创建 RDD 的过程我们可以认为没有 task 的概念，比如 读取 HDFS 文件；

有了 RDD 后才有 task 的概念； 

重点

一个 inputSplit 对应 RDD 的一个 partition；

RDD 的一个 partition 对应一个 task，也就是说 一个 inputSplit 对应一个 task；

通常情况下 一个 block 对应一个 inputSplit；

　　// 以 textFile 为例，每个 inputSplit 不能大于 blockSize，也就是说 可以把 block 切开，但不能把多个 block 组合起来，如果不指定分区，那么每个切片就是 block；

作如下实验证明上述结论

import time
from pyspark import SparkContext

time.clock()
sc = SparkContext(master='yarn')
rdd = sc.textFile('/spark/gps/GPS3.csv', 2).repartition(100).map(lambda x: x).count()
print(time.clock())

##### GPS3.csv 315M，分为 3 个 block
#### 不指定分区-100 runtime：0.64
### 划分2个 stage，
# 第一个 stage sc.textFile('/spark/gps/GPS3.csv').repartition(100) 共 3 个task，
# 第二个 stage .map(lambda x: x).count() 共 100个task
# 19/12/10 22:16:15 INFO cluster.YarnScheduler: Adding task set 0.0 with 3 tasks
# 19/12/10 22:16:34 INFO scheduler.TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0, hadoop13, executor 2, partition 0, NODE_LOCAL, 7899 bytes)
# 19/12/10 22:17:07 INFO cluster.YarnScheduler: Adding task set 1.0 with 100 tasks

#### 指定 5 个分区-100 runtime：0.54
### 划分2个 stage，
# 第一个 stage sc.textFile('/spark/gps/GPS3.csv').repartition(100) 共 5 个task，
# 第二个 stage .map(lambda x: x).count() 共 100个task
# 19/12/10 22:23:09 INFO cluster.YarnScheduler: Adding task set 0.0 with 5 tasks
# 19/12/10 22:17:07 INFO cluster.YarnScheduler: Adding task set 1.0 with 100 tasks

#### 指定 2 个分区-100 runtime：0.6
### 划分2个 stage，
# 第一个 stage sc.textFile('/spark/gps/GPS3.csv').repartition(100) 共 3 个task，
# 第二个 stage .map(lambda x: x).count() 共 100个task
# 19/12/10 22:23:09 INFO cluster.YarnScheduler: Adding task set 0.0 with 3 tasks
# 19/12/10 22:17:07 INFO cluster.YarnScheduler: Adding task set 1.0 with 100 tasks

可以看到

task 并行度

首先明确一点，并行度与 task 数并无关系，并行度是由 spark-submit 提交的参数决定的

taskSet 被分发到多个 Executor 执行；

每个节点可以运行多个 Executor，一个 Executor 相当于一个进程；

一个 Executor 可以有多个 core，一个 core 执行一个 task，一个 core 相当于 Executor 进程里的一个线程；

task 的并发度 = Executor 数 x core 数 = 总 core 数；

对应到 yarn 模式的 spark-submit 参数

--num-executors

--executor-cores

--total-executor-cores　　【这个参数官方解释只能用于 Spark standalone and Mesos only 模式，不过我用在 yarn 模式没报错】

试想如果有 100 个任务，20 个 Executor，每个 Executor 5 个 core，那么资源利用率极高；

然而加入只有 10 个任务，还是 20 个 Executor，每个 Executor 5 个 core，那么资源有很大浪费，这是 spark 调优的一个方向

参考资料：

https://blog.csdn.net/u012965373/article/details/80847543

https://blog.csdn.net/abc_321a/article/details/82020974