Spark.speculation詳解

在大數據處理場景中，Spark一直是一個被廣泛使用的框架。對於Spark性能優化的探索也一直不停止。Spark.speculation是其中一個性能優化的重要手段之一。本文將圍繞著Spark.speculation來進行展開。

一、啟用Spark.speculation=true

首先，讓我們來看一下如何啟用Spark.speculation。通過設置SparkConf中的spark.speculation參數來啟用：

SparkConf sparkConf =  new SparkConf().setAppName(appName)
.set("spark.speculation", "true");

這可能是最簡單的啟用Spark.speculation的方法。如果您需要更多的Spark.speculation定製參數，請參閱下一節。

二、Spark.speculation參數說明

1. spark.speculation.interval

spark.speculation.interval表示兩次檢查確認殘留任務的間隔時間。默認情況下，此參數設置為100毫秒。

值得注意的是，如果您設置此值過高，則可能會減緩Spark中的任務完成。因此，在特定環境下，您可能需要將此參數視為最佳設置。

SparkConf sparkConf = new SparkConf().setAppName(appName)
.set("spark.speculation", "true")
.set("spark.speculation.interval", "50ms");

2. spark.speculation.multiplier

spark.speculation.multiplier 用於計算任務是否被認為是慢任務。該參數的默認值為1.5。因此，如果計算任務已經超過該任務的平均時間的1.5倍，則認為該任務是一個慢任務。

如果您認為設置1.5的倍增因子不足以檢測到某些特定運行緩慢的任務，那麼可以適當增加此倍增因子。

SparkConf sparkConf = new SparkConf().setAppName(appName)
.set("spark.speculation", "true")
.set("spark.speculation.multiplier", "2.0");

3. spark.speculation.quantile（已棄用）

該參數原本存放了任務完成時間的百分位數，以便確定任務的執行時間。但是，從Spark 2.0.0版本開始，該參數已經被棄用並被spark.speculation.multiplier取代。

三、實戰應用Spark.speculation

1. 按 Spark.speculation默認配置運行任務

在示例代碼中，我們將使用如下Spark任務：

JavaRDD rdd = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9))
...
JavaPairRDD<Integer, Integer> pairRDD = rdd.mapToPair(i -> new Tuple2(i, i * 2));
...
JavaPairRDD<Integer, Integer> resultRDD = pairRDD.reduceByKey((x,y)->x+y);
resultRDD.foreach(x -> System.out.println(x._1() + ':' + x._2()));

在默認情況下，Spark不會啟用Spark speculation。因此，任務完成時間可能會非常長，如下圖所示：

在本例中，*任務8* 可能是我們需要解決的問題。在Spark的任務日誌中，我們可以看到，任務8的執行時間是6819毫秒，這遠遠高於任務的平均執行時間。這表明任務8正在緩慢運行。

2. 啟用Spark speculations

為了啟用Spark speculations，在代碼中設置SparkConf對象即可。

SparkConf sparkConf =  new SparkConf().setAppName(appName)
.set("spark.speculation", "true");
JavaSparkContext sc = new JavaSparkContext(sparkConf);

設置Spark.speculation以後，再次運行任務：

JavaRDD rdd = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9))
...
JavaPairRDD<Integer, Integer> pairRDD = rdd.mapToPair(i -> new Tuple2(i, i * 2));
...
JavaPairRDD<Integer, Integer> resultRDD = pairRDD.reduceByKey((x,y)->x+y);
resultRDD.foreach(x -> System.out.println(x._1() + ':' + x._2()));

此時，在那些運行緩慢的任務上計算的機器將啟用Spark speculation進一步計算結果並驗證結果正確性

最終結果如下所示：

我們看到任務 8 的執行時間大大縮短，這意味著Spark.speculation在該任務上有效工作。

四、總結

通過Spark.speculation，您可以輕鬆地檢測緩慢運行的任務並重新計算以提高任務執行時間和Spark作業的整體執行效率。通過本文的介紹，相信您能更好的了解Spark.speculation的原理和實際應用。