Spark调优之推测执行

基本概念

在spark程序中，推测任务是指对于一个stage里面拖后腿的task，会在其他节点的Executor上再次启动这个task，如果其中一个task的实例运行成功者将这个最先完成的task的计算结果，同时会干掉其它Executor上运行的实例。默认情况下推测执行时关闭的。

开启推测的优点：

解决慢task提升作业的整体执行进度
解决分布式集群环境下，负载不均衡或者资源分布不均等问题
解决因机器或者程序bug导致执行task的进程hang（暂时停止执行）住，使得job无法继续执行，需要重启任务等问题

开启推测的弊端：

占用更多的集群资源，严重的会造成所有资源被全部占用，不能及时释放
task执行非事务操作，如果中间过程有跟外界存储交互的可能会影响结果数据

推测执行算法流程图：

开启spark的推测执行，需要设置运行参数spark.speculation=true，两种设置方式:

在程序的sparkConf对象设置：sparkConf.set("spark.speculation","true")
提交作业时设定： --conf spark.speculation=true

开启spark的推测执行需结合其他三个参数同时使用：

spark.speculation.interval 100：检测周期，单位毫秒；
spark.speculation.quantile 0.75：完成task的百分比时启动推测；
spark.speculation.multiplier 1.5：比其他的慢多少倍时启动推测。

执行流程如下图：

执行流程：推测执行根据设置检查周期spark.speculation.interval，默认100ms定时检查执行的task是否需要对task启用推测。当task执行到100ms时，程序开始检测该spark程序job对应的stage已经执行完成的task，如果没有超过spark.speculation.quantile设定的百分比，则不启用推测。如果超过spark.speculation.quantile设定的值，计算成功task运行时间的中位数medianDuration，然后计算启用推测执行时间的界限threshold = （spark.speculation.multiplier）*medianDuration，对正在运行的task运行时间是否超过启用推测执行时间的界限threshold，如果运行时间未超过界限，则不启用推测，如果超过界限则会在另一个excecutor启动相同的task计算，如果其中一个task的实例运行成功者将这个最先完成的task的计算结果，同时会干掉其它Executor上运行的实例。如果200ms的时候，也就是spark.speculation.interval的2倍还有task未完成的话，就会进入下一次的推测执行判断周期中，判断逻辑跟周期一的一样，这是一个循环的过程。

注意问题：

推测执行的检测周期不要设计得太短，不然可能会重复创建很多相同的task，如果有实时跟外部存储交互的场景慎用推测执行，因为一个task虽然没有执行完，但是一部分结果已经写入外部存储了，启动多个task就会造成数据重复，所以具体要不要开推测和参数怎么设定，一定要根据具体业务设定。

参考：

https://www.2cto.com/kf/201803/734284.html

https://blog.csdn.net/lvbiao_62/article/details/79751560