為何要處理數據傾斜（Data Skew）

什么是數據傾斜

對Spark/Hadoop這樣的大數據系統來講，數據量大并不可怕，可怕的是數據傾斜。

何謂數據傾斜？數據傾斜指的是，并行處理的數據集中，某一部分（如Spark或Kafka的一個Partition）的數據顯著多于其它部分，從而使得該部分的處理速度成為整個數據集處理的瓶頸。

數據傾斜是如何造成的

 在Spark中，同一個Stage的不同Partition可以并行處理，而具有依賴關系的不同Stage之間是串行處理的。假設某個Spark Job分為Stage 0和Stage 1兩個Stage，且Stage 1依賴于Stage 0，那Stage 0完全處理結束之前不會處理Stage 1。而Stage 0可能包含N個Task，這N個Task可以并行進行。如果其中N-1個Task都在10秒內完成，而另外一個Task卻耗時1分鐘，那該Stage的總時間至少為1分鐘。換句話說，一個Stage所耗費的時間，主要由最慢的那個Task決定。

由于同一個Stage內的所有Task執行相同的計算，在排除不同計算節點計算能力差異的前提下，不同Task之間耗時的差異主要由該Task所處理的數據量決定。

Stage的數據來源主要分為如下兩類

• 從數據源直接讀取。如讀取HDFS，Kafka
• 讀取上一個Stage的Shuffle數據

如何緩解/消除數據傾斜

盡量避免數據源的數據傾斜

以Spark Stream通過DirectStream方式讀取Kafka數據為例。由于Kafka的每一個Partition對應Spark的一個Task（Partition），所以Kafka內相關Topic的各Partition之間數據是否平衡，直接決定Spark處理該數據時是否會產生數據傾斜。

如《Kafka設計解析（一）- Kafka背景及架構介紹》一文所述，Kafka某一Topic內消息在不同Partition之間的分布，主要由Producer端所使用的Partition實現類決定。如果使用隨機Partitioner，則每條消息會隨機發送到一個Partition中，從而從概率上來講，各Partition間的數據會達到平衡。此時源Stage（直接讀取Kafka數據的Stage）不會產生數據傾斜。

但很多時候，業務場景可能會要求將具備同一特征的數據順序消費，此時就需要將具有相同特征的數據放于同一個Partition中。一個典型的場景是，需要將同一個用戶相關的PV信息置于同一個Partition中。此時，如果產生了數據傾斜，則需要通過其它方式處理。

調整并行度分散同一個Task的不同Key

原理

Spark在做Shuffle時，默認使用HashPartitioner（非Hash Shuffle）對數據進行分區。如果并行度設置的不合適，可能造成大量不相同的Key對應的數據被分配到了同一個Task上，造成該Task所處理的數據遠大于其它Task，從而造成數據傾斜。

 如果調整Shuffle時的并行度，使得原本被分配到同一Task的不同Key發配到不同Task上處理，則可降低原Task所需處理的數據量，從而緩解數據傾斜問題造成的短板效應。

案例

現有一張測試表，名為student_external，內有10.5億條數據，每條數據有一個唯一的id值。現從中取出id取值為9億到10.5億的共1.5條數據，并通過一些處理，使得id為9億到9.4億間的所有數據對12取模后余數為8（即在Shuffle并行度為12時該數據集全部被HashPartition分配到第8個Task），其它數據集對其id除以100取整，從而使得id大于9.4億的數據在Shuffle時可被均勻分配到所有Task中，而id小于9.4億的數據全部分配到同一個Task中。處理過程如下

INSERT OVERWRITE TABLE test
SELECT CASE WHEN id < 940000000 THEN (9500000  + (CAST (RAND() * 8 AS INTEGER)) * 12 )
       ELSE CAST(id/100 AS INTEGER)
       END,
       name
FROM student_external
WHERE id BETWEEN 900000000 AND 1050000000;

通過上述處理，一份可能造成后續數據傾斜的測試數據即以準備好。接下來，使用Spark讀取該測試數據，并通過groupByKey(12)對id分組處理，且Shuffle并行度為12。代碼如下

public class SparkDataSkew {
  public static void main(String[] args) {
    SparkSession sparkSession = SparkSession.builder()
      .appName("SparkDataSkewTunning")
      .config("hive.metastore.uris", "thrift://hadoop1:9083")
      .enableHiveSupport()
      .getOrCreate();

    Dataset dataframe = sparkSession.sql( "select * from test");
    dataframe.toJavaRDD()
      .mapToPair((Row row) -> new Tuple2(row.getInt(0),row.getString(1)))
      .groupByKey(12)
      .mapToPair((Tuple2> tuple) -> {
        int id = tuple._1();
        AtomicInteger atomicInteger = new AtomicInteger(0);
        tuple._2().forEach((String name) -> atomicInteger.incrementAndGet());
        return new Tuple2(id, atomicInteger.get());
      }).count();

      sparkSession.stop();
      sparkSession.close();
  }
  
}

本次實驗所使用集群節點數為4，每個節點可被Yarn使用的CPU核數為16，內存為16GB。使用如下方式提交上述應用，將啟動4個Executor，每個Executor可使用核數為12（該配置并非生產環境下的最優配置，僅用于本文實驗），可用內存為12GB。

spark-submit --queue ambari --num-executors 4 --executor-cores 12 --executor-memory 12g --class com.jasongj.spark.driver.SparkDataSkew --master yarn --deploy-mode client SparkExample-with-dependencies-1.0.jar

GroupBy Stage的Task狀態如下圖所示，Task 8處理的記錄數為4500萬，遠大于（9倍于）其它11個Task處理的500萬記錄。而Task 8所耗費的時間為38秒，遠高于其它11個Task的平均時間（16秒）。整個Stage的時間也為38秒，該時間主要由最慢的Task 8決定。

在這種情況下，可以通過調整Shuffle并行度，使得原來被分配到同一個Task（即該例中的Task 8）的不同Key分配到不同Task，從而降低Task 8所需處理的數據量，緩解數據傾斜。

通過groupByKey(48)將Shuffle并行度調整為48，重新提交到Spark。新的Job的GroupBy Stage所有Task狀態如下圖所示。

從上圖可知，記錄數最多的Task 20處理的記錄數約為1125萬，相比于并行度為12時Task 8的4500萬，降低了75%左右，而其耗時從原來Task 8的38秒降到了24秒。

 在這種場景下，調整并行度，并不意味著一定要增加并行度，也可能是減小并行度。如果通過groupByKey(11)將Shuffle并行度調整為11，重新提交到Spark。新Job的GroupBy Stage的所有Task狀態如下圖所示。

從上圖可見，處理記錄數最多的Task 6所處理的記錄數約為1045萬，耗時為23秒。處理記錄數最少的Task 1處理的記錄數約為545萬，耗時12秒。

總結

適用場景
大量不同的Key被分配到了相同的Task造成該Task數據量過大。

解決方案
調整并行度。一般是增大并行度，但有時如本例減小并行度也可達到效果。

優勢
實現簡單，可在需要Shuffle的操作算子上直接設置并行度或者使用spark.default.parallelism設置。如果是Spark SQL，還可通過SET spark.sql.shuffle.partitions=[num_tasks]設置并行度。可用最小的代價解決問題。一般如果出現數據傾斜，都可以通過這種方法先試驗幾次，如果問題未解決，再嘗試其它方法。

劣勢
適用場景少，只能將分配到同一Task的不同Key分散開，但對于同一Key傾斜嚴重的情況該方法并不適用。并且該方法一般只能緩解數據傾斜，沒有徹底消除問題。從實踐經驗來看，其效果一般。

自定義Partitioner

原理

使用自定義的Partitioner（默認為HashPartitioner），將原本被分配到同一個Task的不同Key分配到不同Task。

案例

以上述數據集為例，繼續將并發度設置為12，但是在groupByKey算子上，使用自定義的Partitioner（實現如下）

.groupByKey(new Partitioner() {
  @Override
  public int numPartitions() {
    return 12;
  }

  @Override
  public int getPartition(Object key) {
    int id = Integer.parseInt(key.toString());
    if(id >= 9500000 && id <= 9500084 && ((id - 9500000) % 12) == 0) {
      return (id - 9500000) / 12;
    } else {
      return id % 12;
    }
  }
})

由下圖可見，使用自定義Partition后，耗時最長的Task 6處理約1000萬條數據，用時15秒。并且各Task所處理的數據集大小相當。

總結

適用場景
大量不同的Key被分配到了相同的Task造成該Task數據量過大。

解決方案
使用自定義的Partitioner實現類代替默認的HashPartitioner，盡量將所有不同的Key均勻分配到不同的Task中。

優勢
不影響原有的并行度設計。如果改變并行度，后續Stage的并行度也會默認改變，可能會影響后續Stage。

劣勢
適用場景有限，只能將不同Key分散開，對于同一Key對應數據集非常大的場景不適用。效果與調整并行度類似，只能緩解數據傾斜而不能完全消除數據傾斜。而且需要根據數據特點自定義專用的Partitioner，不夠靈活。

將Reduce side Join轉變為Map side Join

原理通過Spark的Broadcast機制，將Reduce側Join轉化為Map側Join，避免Shuffle從而完全消除Shuffle帶來的數據傾斜。

案例

通過如下SQL創建一張具有傾斜Key且總記錄數為1.5億的大表test。

INSERT OVERWRITE TABLE test
SELECT CAST(CASE WHEN id < 980000000 THEN (95000000  + (CAST (RAND() * 4 AS INT) + 1) * 48 )
       ELSE CAST(id/10 AS INT) END AS STRING),
       name
FROM student_external
WHERE id BETWEEN 900000000 AND 1050000000;

使用如下SQL創建一張數據分布均勻且總記錄數為50萬的小表test_new。

INSERT OVERWRITE TABLE test_new
SELECT CAST(CAST(id/10 AS INT) AS STRING),
       name
FROM student_delta_external
WHERE id BETWEEN 950000000 AND 950500000;

直接通過Spark Thrift Server提交如下SQL將表test與表test_new進行Join并將Join結果存于表test_join中。

INSERT OVERWRITE TABLE test_join
SELECT test_new.id, test_new.name
FROM test
JOIN test_new
ON test.id = test_new.id;

該SQL對應的DAG如下圖所示。從該圖可見，該執行過程總共分為三個Stage，前兩個用于從Hive中讀取數據，同時二者進行Shuffle，通過最后一個Stage進行Join并將結果寫入表test_join中。

從下圖可見，最近Join Stage各Task處理的數據傾斜嚴重，處理數據量最大的Task耗時7.1分鐘，遠高于其它無數據傾斜的Task約2s秒的耗時。

 接下來，嘗試通過Broadcast實現Map側Join。實現Map側Join的方法，并非直接通過CACHE TABLE test_new將小表test_new進行cache。現通過如下SQL進行Join。

CACHE TABLE test_new;
INSERT OVERWRITE TABLE test_join
SELECT test_new.id, test_new.name
FROM test
JOIN test_new
ON test.id = test_new.id;

SET spark.sql.autoBroadcastJoinThreshold=104857600;
INSERT OVERWRITE TABLE test_join
SELECT test_new.id, test_new.name
FROM test
JOIN test_new
ON test.id = test_new.id;