Spark性能調優-高級篇

時間  2020-05-28
標籤 spark 性能 高級 欄目 Spark 简体版
原文   原文鏈接

前言

繼基礎篇講解了每一個Spark開發人員都必須熟知的開發調優與資源調優以後，本文做爲《Spark性能優化指南》的高級篇，將深刻分析數據傾斜調優與shuffle調優，以解決更加棘手的性能問題。html

數據傾斜調優

調優概述

有的時候，咱們可能會遇到大數據計算中一個最棘手的問題——數據傾斜，此時Spark做業的性能會比指望差不少。數據傾斜調優，就是使用各類技術方案解決不一樣類型的數據傾斜問題，以保證Spark做業的性能。算法

數據傾斜發生時的現象

  • 絕大多數task執行得都很是快，但個別task執行極慢。好比，總共有1000個task，997個task都在1分鐘以內執行完了，可是剩餘兩三個task卻要一兩個小時。這種狀況很常見。sql

  • 本來可以正常執行的Spark做業，某天忽然報出OOM（內存溢出）異常，觀察異常棧，是咱們寫的業務代碼形成的。這種狀況比較少見。後端

數據傾斜發生的原理

數據傾斜的原理很簡單：在進行shuffle的時候，必須將各個節點上相同的key拉取到某個節點上的一個task來進行處理，好比按照key進行聚合或join等操做。此時若是某個key對應的數據量特別大的話，就會發生數據傾斜。好比大部分key對應10條數據，可是個別key卻對應了100萬條數據，那麼大部分task可能就只會分配到10條數據，而後1秒鐘就運行完了；可是個別task可能分配到了100萬數據，要運行一兩個小時。所以，整個Spark做業的運行進度是由運行時間最長的那個task決定的。性能優化

所以出現數據傾斜的時候，Spark做業看起來會運行得很是緩慢，甚至可能由於某個task處理的數據量過大致使內存溢出。網絡

舉一個很清晰的例子：hello這個key，在三個節點上對應了總共7條數據，這些數據都會被拉取到同一個task中進行處理；而world和you這兩個key分別纔對應1條數據，因此另外兩個task只要分別處理1條數據便可。此時第一個task的運行時間多是另外兩個task的7倍，而整個stage的運行速度也由運行最慢的那個task所決定。數據結構

 

如何定位致使數據傾斜的代碼

數據傾斜只會發生在shuffle過程當中。這裏給你們羅列一些經常使用的而且可能會觸發shuffle操做的算子：distinct、groupByKey、reduceByKey、aggregateByKey、join、cogroup、repartition等。出現數據傾斜時，可能就是你的代碼中使用了這些算子中的某一個所致使的。app

某個task執行特別慢的狀況

首先要看的，就是數據傾斜發生在第幾個stage中。dom

若是是用yarn-client模式提交，那麼本地是直接能夠看到log的，能夠在log中找到當前運行到了第幾個stage；若是是用yarn-cluster模式提交，則能夠經過Spark Web UI來查看當前運行到了第幾個stage。此外，不管是使用yarn-client模式仍是yarn-cluster模式，咱們均可以在Spark Web UI上深刻看一下當前這個stage各個task分配的數據量，從而進一步肯定是否是task分配的數據不均勻致使了數據傾斜。ide

好比下圖中，倒數第三列顯示了每一個task的運行時間。明顯能夠看到，有的task運行特別快，只須要幾秒鐘就能夠運行完；而有的task運行特別慢，須要幾分鐘才能運行完，此時單從運行時間上看就已經可以肯定發生數據傾斜了。此外，倒數第一列顯示了每一個task處理的數據量，明顯能夠看到，運行時間特別短的task只須要處理幾百KB的數據便可，而運行時間特別長的task須要處理幾千KB的數據，處理的數據量差了10倍。此時更加可以肯定是發生了數據傾斜。

知道數據傾斜發生在哪個stage以後，接着咱們就須要根據stage劃分原理，推算出來發生傾斜的那個stage對應代碼中的哪一部分，這部分代碼中確定會有一個shuffle類算子。精準推算stage與代碼的對應關係，須要對Spark的源碼有深刻的理解，這裏咱們能夠介紹一個相對簡單實用的推算方法：只要看到Spark代碼中出現了一個shuffle類算子或者是Spark SQL的SQL語句中出現了會致使shuffle的語句（好比group by語句），那麼就能夠斷定，以那個地方爲界限劃分出了先後兩個stage。

這裏咱們就以Spark最基礎的入門程序——單詞計數來舉例，如何用最簡單的方法大體推算出一個stage對應的代碼。以下示例，在整個代碼中，只有一個reduceByKey是會發生shuffle的算子，所以就能夠認爲，以這個算子爲界限，會劃分出先後兩個stage。

  • stage0，主要是執行從textFile到map操做，以及執行shuffle write操做。shuffle write操做，咱們能夠簡單理解爲對pairs RDD中的數據進行分區操做，每一個task處理的數據中，相同的key會寫入同一個磁盤文件內。
  • stage1，主要是執行從reduceByKey到collect操做，stage1的各個task一開始運行，就會首先執行shuffle read操做。執行shuffle read操做的task，會從stage0的各個task所在節點拉取屬於本身處理的那些key，而後對同一個key進行全局性的聚合或join等操做，在這裏就是對key的value值進行累加。stage1在執行完reduceByKey算子以後，就計算出了最終的wordCounts RDD，而後會執行collect算子，將全部數據拉取到Driver上，供咱們遍歷和打印輸出。
val conf = new SparkConf()
val sc = new SparkContext(conf)

val lines = sc.textFile("hdfs://...")
val words = lines.flatMap(_.split(" "))
val pairs = words.map((_, 1))
val wordCounts = pairs.reduceByKey(_ + _)

wordCounts.collect().foreach(println(_))

經過對單詞計數程序的分析，但願可以讓你們瞭解最基本的stage劃分的原理，以及stage劃分後shuffle操做是如何在兩個stage的邊界處執行的。而後咱們就知道如何快速定位出發生數據傾斜的stage對應代碼的哪個部分了。好比咱們在Spark Web UI或者本地log中發現，stage1的某幾個task執行得特別慢，斷定stage1出現了數據傾斜，那麼就能夠回到代碼中定位出stage1主要包括了reduceByKey這個shuffle類算子，此時基本就能夠肯定是由educeByKey算子致使的數據傾斜問題。好比某個單詞出現了100萬次，其餘單詞纔出現10次，那麼stage1的某個task就要處理100萬數據，整個stage的速度就會被這個task拖慢。

某個task莫名其妙內存溢出的狀況

這種狀況下去定位出問題的代碼就比較容易了。咱們建議直接看yarn-client模式下本地log的異常棧，或者是經過YARN查看yarn-cluster模式下的log中的異常棧。通常來講，經過異常棧信息就能夠定位到你的代碼中哪一行發生了內存溢出。而後在那行代碼附近找找，通常也會有shuffle類算子，此時極可能就是這個算子致使了數據傾斜。

可是你們要注意的是，不能單純靠偶然的內存溢出就斷定發生了數據傾斜。由於本身編寫的代碼的bug，以及偶然出現的數據異常，也可能會致使內存溢出。所以仍是要按照上面所講的方法，經過Spark Web UI查看報錯的那個stage的各個task的運行時間以及分配的數據量，才能肯定是不是因爲數據傾斜才致使了此次內存溢出。

查看致使數據傾斜的key的數據分佈狀況

知道了數據傾斜發生在哪裏以後，一般須要分析一下那個執行了shuffle操做而且致使了數據傾斜的RDD/Hive表，查看一下其中key的分佈狀況。這主要是爲以後選擇哪種技術方案提供依據。針對不一樣的key分佈與不一樣的shuffle算子組合起來的各類狀況，可能須要選擇不一樣的技術方案來解決。

此時根據你執行操做的狀況不一樣，能夠有不少種查看key分佈的方式：

  1. 若是是Spark SQL中的group by、join語句致使的數據傾斜，那麼就查詢一下SQL中使用的表的key分佈狀況。
  2. 若是是對Spark RDD執行shuffle算子致使的數據傾斜，那麼能夠在Spark做業中加入查看key分佈的代碼，好比RDD.countByKey()。而後對統計出來的各個key出現的次數，collect/take到客戶端打印一下，就能夠看到key的分佈狀況。

舉例來講，對於上面所說的單詞計數程序，若是肯定了是stage1的reduceByKey算子致使了數據傾斜，那麼就應該看看進行reduceByKey操做的RDD中的key分佈狀況，在這個例子中指的就是pairs RDD。以下示例，咱們能夠先對pairs採樣10%的樣本數據，而後使用countByKey算子統計出每一個key出現的次數，最後在客戶端遍歷和打印樣本數據中各個key的出現次數。

val sampledPairs = pairs.sample(false, 0.1)
val sampledWordCounts = sampledPairs.countByKey()
sampledWordCounts.foreach(println(_))

數據傾斜的解決方案

解決方案一：使用Hive ETL預處理數據

方案適用場景：致使數據傾斜的是Hive表。若是該Hive表中的數據自己很不均勻（好比某個key對應了100萬數據，其餘key纔對應了10條數據），並且業務場景須要頻繁使用Spark對Hive表執行某個分析操做，那麼比較適合使用這種技術方案。

方案實現思路：此時能夠評估一下，是否能夠經過Hive來進行數據預處理（即經過Hive ETL預先對數據按照key進行聚合，或者是預先和其餘表進行join），而後在Spark做業中針對的數據源就不是原來的Hive表了，而是預處理後的Hive表。此時因爲數據已經預先進行過聚合或join操做了，那麼在Spark做業中也就不須要使用原先的shuffle類算子執行這類操做了。

方案實現原理：這種方案從根源上解決了數據傾斜，由於完全避免了在Spark中執行shuffle類算子，那麼確定就不會有數據傾斜的問題了。可是這裏也要提醒一下你們，這種方式屬於治標不治本。由於畢竟數據自己就存在分佈不均勻的問題，因此Hive ETL中進行group by或者join等shuffle操做時，仍是會出現數據傾斜，致使Hive ETL的速度很慢。咱們只是把數據傾斜的發生提早到了Hive ETL中，避免Spark程序發生數據傾斜而已。

方案優勢：實現起來簡單便捷，效果還很是好，徹底規避掉了數據傾斜，Spark做業的性能會大幅度提高。

方案缺點：治標不治本，Hive ETL中仍是會發生數據傾斜。

方案實踐經驗：在一些Java系統與Spark結合使用的項目中，會出現Java代碼頻繁調用Spark做業的場景，並且對Spark做業的執行性能要求很高，就比較適合使用這種方案。將數據傾斜提早到上游的Hive ETL，天天僅執行一次，只有那一次是比較慢的，而以後每次Java調用Spark做業時，執行速度都會很快，可以提供更好的用戶體驗。

項目實踐經驗：在美團·點評的交互式用戶行爲分析系統中使用了這種方案，該系統主要是容許用戶經過Java Web系統提交數據分析統計任務，後端經過Java提交Spark做業進行數據分析統計。要求Spark做業速度必需要快，儘可能在10分鐘之內，不然速度太慢，用戶體驗會不好。因此咱們將有些Spark做業的shuffle操做提早到了Hive ETL中，從而讓Spark直接使用預處理的Hive中間表，儘量地減小Spark的shuffle操做，大幅度提高了性能，將部分做業的性能提高了6倍以上。

解決方案二：過濾少數致使傾斜的key

方案適用場景：若是發現致使傾斜的key就少數幾個，並且對計算自己的影響並不大的話，那麼很適合使用這種方案。好比99%的key就對應10條數據，可是隻有一個key對應了100萬數據，從而致使了數據傾斜。

方案實現思路：若是咱們判斷那少數幾個數據量特別多的key，對做業的執行和計算結果不是特別重要的話，那麼幹脆就直接過濾掉那少數幾個key。好比，在Spark SQL中可使用where子句過濾掉這些key或者在Spark Core中對RDD執行filter算子過濾掉這些key。若是須要每次做業執行時，動態斷定哪些key的數據量最多而後再進行過濾，那麼可使用sample算子對RDD進行採樣，而後計算出每一個key的數量，取數據量最多的key過濾掉便可。

方案實現原理：將致使數據傾斜的key給過濾掉以後，這些key就不會參與計算了，天然不可能產生數據傾斜。

方案優勢：實現簡單，並且效果也很好，能夠徹底規避掉數據傾斜。

方案缺點：適用場景很少，大多數狀況下，致使傾斜的key仍是不少的，並非只有少數幾個。

方案實踐經驗：在項目中咱們也採用過這種方案解決數據傾斜。有一次發現某一天Spark做業在運行的時候忽然OOM了，追查以後發現，是Hive表中的某一個key在那天數據異常，致使數據量暴增。所以就採起每次執行前先進行採樣，計算出樣本中數據量最大的幾個key以後，直接在程序中將那些key給過濾掉。

解決方案三：提升shuffle操做的並行度

方案適用場景：若是咱們必需要對數據傾斜迎難而上，那麼建議優先使用這種方案，由於這是處理數據傾斜最簡單的一種方案。

方案實現思路：在對RDD執行shuffle算子時，給shuffle算子傳入一個參數，好比reduceByKey(1000)，該參數就設置了這個shuffle算子執行時shuffle read task的數量。對於Spark SQL中的shuffle類語句，好比group by、join等，須要設置一個參數，即spark.sql.shuffle.partitions，該參數表明了shuffle read task的並行度，該值默認是200，對於不少場景來講都有點太小。

方案實現原理：增長shuffle read task的數量，可讓本來分配給一個task的多個key分配給多個task，從而讓每一個task處理比原來更少的數據。舉例來講，若是本來有5個key，每一個key對應10條數據，這5個key都是分配給一個task的，那麼這個task就要處理50條數據。而增長了shuffle read task之後，每一個task就分配到一個key，即每一個task就處理10條數據，那麼天然每一個task的執行時間都會變短了。具體原理以下圖所示。

方案優勢：實現起來比較簡單，能夠有效緩解和減輕數據傾斜的影響。

方案缺點：只是緩解了數據傾斜而已，沒有完全根除問題，根據實踐經驗來看，其效果有限。

方案實踐經驗：該方案一般沒法完全解決數據傾斜，由於若是出現一些極端狀況，好比某個key對應的數據量有100萬，那麼不管你的task數量增長到多少，這個對應着100萬數據的key確定仍是會分配到一個task中去處理，所以註定仍是會發生數據傾斜的。因此這種方案只能說是在發現數據傾斜時嘗試使用的第一種手段，嘗試去用嘴簡單的方法緩解數據傾斜而已，或者是和其餘方案結合起來使用。

解決方案四：兩階段聚合（局部聚合+全局聚合）

方案適用場景：對RDD執行reduceByKey等聚合類shuffle算子或者在Spark SQL中使用group by語句進行分組聚合時，比較適用這種方案。

方案實現思路：這個方案的核心實現思路就是進行兩階段聚合。第一次是局部聚合，先給每一個key都打上一個隨機數，好比10之內的隨機數，此時原先同樣的key就變成不同的了，好比(hello, 1) (hello, 1) (hello, 1) (hello, 1)，就會變成(1_hello, 1) (1_hello, 1) (2_hello, 1) (2_hello, 1)。接着對打上隨機數後的數據，執行reduceByKey等聚合操做，進行局部聚合，那麼局部聚合結果，就會變成了(1_hello, 2) (2_hello, 2)。而後將各個key的前綴給去掉，就會變成(hello,2)(hello,2)，再次進行全局聚合操做，就能夠獲得最終結果了，好比(hello, 4)。

方案實現原理：將本來相同的key經過附加隨機前綴的方式，變成多個不一樣的key，就可讓本來被一個task處理的數據分散到多個task上去作局部聚合，進而解決單個task處理數據量過多的問題。接着去除掉隨機前綴，再次進行全局聚合，就能夠獲得最終的結果。具體原理見下圖。

方案優勢：對於聚合類的shuffle操做致使的數據傾斜，效果是很是不錯的。一般均可以解決掉數據傾斜，或者至少是大幅度緩解數據傾斜，將Spark做業的性能提高數倍以上。

方案缺點：僅僅適用於聚合類的shuffle操做，適用範圍相對較窄。若是是join類的shuffle操做，還得用其餘的解決方案。

// 第一步，給RDD中的每一個key都打上一個隨機前綴。
JavaPairRDD<String, Long> randomPrefixRdd = rdd.mapToPair(
        new PairFunction<Tuple2<Long,Long>, String, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<String, Long> call(Tuple2<Long, Long> tuple)
                    throws Exception {
                Random random = new Random();
                int prefix = random.nextInt(10);
                return new Tuple2<String, Long>(prefix + "_" + tuple._1, tuple._2);
            }
        });

// 第二步，對打上隨機前綴的key進行局部聚合。
JavaPairRDD<String, Long> localAggrRdd = randomPrefixRdd.reduceByKey(
        new Function2<Long, Long, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Long call(Long v1, Long v2) throws Exception {
                return v1 + v2;
            }
        });

// 第三步，去除RDD中每一個key的隨機前綴。
JavaPairRDD<Long, Long> removedRandomPrefixRdd = localAggrRdd.mapToPair(
        new PairFunction<Tuple2<String,Long>, Long, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<Long, Long> call(Tuple2<String, Long> tuple)
                    throws Exception {
                long originalKey = Long.valueOf(tuple._1.split("_")[1]);
                return new Tuple2<Long, Long>(originalKey, tuple._2);
            }
        });

// 第四步，對去除了隨機前綴的RDD進行全局聚合。
JavaPairRDD<Long, Long> globalAggrRdd = removedRandomPrefixRdd.reduceByKey(
        new Function2<Long, Long, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Long call(Long v1, Long v2) throws Exception {
                return v1 + v2;
            }
        });

解決方案五：將reduce join轉爲map join

方案適用場景：在對RDD使用join類操做，或者是在Spark SQL中使用join語句時，並且join操做中的一個RDD或表的數據量比較小（好比幾百M或者一兩G），比較適用此方案。

方案實現思路：不使用join算子進行鏈接操做，而使用Broadcast變量與map類算子實現join操做，進而徹底規避掉shuffle類的操做，完全避免數據傾斜的發生和出現。將較小RDD中的數據直接經過collect算子拉取到Driver端的內存中來，而後對其建立一個Broadcast變量；接着對另一個RDD執行map類算子，在算子函數內，從Broadcast變量中獲取較小RDD的全量數據，與當前RDD的每一條數據按照鏈接key進行比對，若是鏈接key相同的話，那麼就將兩個RDD的數據用你須要的方式鏈接起來。

方案實現原理：普通的join是會走shuffle過程的，而一旦shuffle，就至關於會將相同key的數據拉取到一個shuffle read task中再進行join，此時就是reduce join。可是若是一個RDD是比較小的，則能夠採用廣播小RDD全量數據+map算子來實現與join一樣的效果，也就是map join，此時就不會發生shuffle操做，也就不會發生數據傾斜。具體原理以下圖所示。

方案優勢：對join操做致使的數據傾斜，效果很是好，由於根本就不會發生shuffle，也就根本不會發生數據傾斜。

方案缺點：適用場景較少，由於這個方案只適用於一個大表和一個小表的狀況。畢竟咱們須要將小表進行廣播，此時會比較消耗內存資源，driver和每一個Executor內存中都會駐留一份小RDD的全量數據。若是咱們廣播出去的RDD數據比較大，好比10G以上，那麼就可能發生內存溢出了。所以並不適合兩個都是大表的狀況。

// 首先將數據量比較小的RDD的數據，collect到Driver中來。
List<Tuple2<Long, Row>> rdd1Data = rdd1.collect()
// 而後使用Spark的廣播功能，將小RDD的數據轉換成廣播變量，這樣每一個Executor就只有一份RDD的數據。
// 能夠儘量節省內存空間，而且減小網絡傳輸性能開銷。
final Broadcast<List<Tuple2<Long, Row>>> rdd1DataBroadcast = sc.broadcast(rdd1Data);

// 對另一個RDD執行map類操做，而再也不是join類操做。
JavaPairRDD<String, Tuple2<String, Row>> joinedRdd = rdd2.mapToPair(
        new PairFunction<Tuple2<Long,String>, String, Tuple2<String, Row>>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<String, Tuple2<String, Row>> call(Tuple2<Long, String> tuple)
                    throws Exception {
                // 在算子函數中，經過廣播變量，獲取到本地Executor中的rdd1數據。
                List<Tuple2<Long, Row>> rdd1Data = rdd1DataBroadcast.value();
                // 能夠將rdd1的數據轉換爲一個Map，便於後面進行join操做。
                Map<Long, Row> rdd1DataMap = new HashMap<Long, Row>();
                for(Tuple2<Long, Row> data : rdd1Data) {
                    rdd1DataMap.put(data._1, data._2);
                }
                // 獲取當前RDD數據的key以及value。
                String key = tuple._1;
                String value = tuple._2;
                // 從rdd1數據Map中，根據key獲取到能夠join到的數據。
                Row rdd1Value = rdd1DataMap.get(key);
                return new Tuple2<String, String>(key, new Tuple2<String, Row>(value, rdd1Value));
            }
        });

// 這裏得提示一下。
// 上面的作法，僅僅適用於rdd1中的key沒有重複，所有是惟一的場景。
// 若是rdd1中有多個相同的key，那麼就得用flatMap類的操做，在進行join的時候不能用map，而是得遍歷rdd1全部數據進行join。
// rdd2中每條數據均可能會返回多條join後的數據。

解決方案六：採樣傾斜key並分拆join操做

方案適用場景：兩個RDD/Hive表進行join的時候，若是數據量都比較大，沒法採用「解決方案五」，那麼此時能夠看一下兩個RDD/Hive表中的key分佈狀況。若是出現數據傾斜，是由於其中某一個RDD/Hive表中的少數幾個key的數據量過大，而另外一個RDD/Hive表中的全部key都分佈比較均勻，那麼採用這個解決方案是比較合適的。

方案實現思路：

  • 對包含少數幾個數據量過大的key的那個RDD，經過sample算子採樣出一份樣原本，而後統計一下每一個key的數量，計算出來數據量最大的是哪幾個key。
  • 而後將這幾個key對應的數據從原來的RDD中拆分出來，造成一個單獨的RDD，並給每一個key都打上n之內的隨機數做爲前綴，而不會致使傾斜的大部分key造成另一個RDD。
  • 接着將須要join的另外一個RDD，也過濾出來那幾個傾斜key對應的數據並造成一個單獨的RDD，將每條數據膨脹成n條數據，這n條數據都按順序附加一個0~n的前綴，不會致使傾斜的大部分key也造成另一個RDD。
  • 再將附加了隨機前綴的獨立RDD與另外一個膨脹n倍的獨立RDD進行join，此時就能夠將原先相同的key打散成n份，分散到多個task中去進行join了。
  • 而另外兩個普通的RDD就照常join便可。
  • 最後將兩次join的結果使用union算子合併起來便可，就是最終的join結果。

方案實現原理：對於join致使的數據傾斜，若是隻是某幾個key致使了傾斜，能夠將少數幾個key分拆成獨立RDD，並附加隨機前綴打散成n份去進行join，此時這幾個key對應的數據就不會集中在少數幾個task上，而是分散到多個task進行join了。具體原理見下圖。

方案優勢：對於join致使的數據傾斜，若是隻是某幾個key致使了傾斜，採用該方式能夠用最有效的方式打散key進行join。並且只須要針對少數傾斜key對應的數據進行擴容n倍，不須要對全量數據進行擴容。避免了佔用過多內存。

方案缺點：若是致使傾斜的key特別多的話，好比成千上萬個key都致使數據傾斜，那麼這種方式也不適合。

// 首先從包含了少數幾個致使數據傾斜key的rdd1中，採樣10%的樣本數據。
JavaPairRDD<Long, String> sampledRDD = rdd1.sample(false, 0.1);

// 對樣本數據RDD統計出每一個key的出現次數，並按出現次數降序排序。
// 對降序排序後的數據，取出top 1或者top 100的數據，也就是key最多的前n個數據。
// 具體取出多少個數據量最多的key，由你們本身決定，咱們這裏就取1個做爲示範。
JavaPairRDD<Long, Long> mappedSampledRDD = sampledRDD.mapToPair(
        new PairFunction<Tuple2<Long,String>, Long, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<Long, Long> call(Tuple2<Long, String> tuple)
                    throws Exception {
                return new Tuple2<Long, Long>(tuple._1, 1L);
            }     
        });
JavaPairRDD<Long, Long> countedSampledRDD = mappedSampledRDD.reduceByKey(
        new Function2<Long, Long, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Long call(Long v1, Long v2) throws Exception {
                return v1 + v2;
            }
        });
JavaPairRDD<Long, Long> reversedSampledRDD = countedSampledRDD.mapToPair( 
        new PairFunction<Tuple2<Long,Long>, Long, Long>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<Long, Long> call(Tuple2<Long, Long> tuple)
                    throws Exception {
                return new Tuple2<Long, Long>(tuple._2, tuple._1);
            }
        });
final Long skewedUserid = reversedSampledRDD.sortByKey(false).take(1).get(0)._2;

// 從rdd1中分拆出致使數據傾斜的key，造成獨立的RDD。
JavaPairRDD<Long, String> skewedRDD = rdd1.filter(
        new Function<Tuple2<Long,String>, Boolean>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Boolean call(Tuple2<Long, String> tuple) throws Exception {
                return tuple._1.equals(skewedUserid);
            }
        });
// 從rdd1中分拆出不致使數據傾斜的普通key，造成獨立的RDD。
JavaPairRDD<Long, String> commonRDD = rdd1.filter(
        new Function<Tuple2<Long,String>, Boolean>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Boolean call(Tuple2<Long, String> tuple) throws Exception {
                return !tuple._1.equals(skewedUserid);
            } 
        });

// rdd2，就是那個全部key的分佈相對較爲均勻的rdd。
// 這裏將rdd2中，前面獲取到的key對應的數據，過濾出來，分拆成單獨的rdd，並對rdd中的數據使用flatMap算子都擴容100倍。
// 對擴容的每條數據，都打上0～100的前綴。
JavaPairRDD<String, Row> skewedRdd2 = rdd2.filter(
         new Function<Tuple2<Long,Row>, Boolean>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Boolean call(Tuple2<Long, Row> tuple) throws Exception {
                return tuple._1.equals(skewedUserid);
            }
        }).flatMapToPair(new PairFlatMapFunction<Tuple2<Long,Row>, String, Row>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Iterable<Tuple2<String, Row>> call(
                    Tuple2<Long, Row> tuple) throws Exception {
                Random random = new Random();
                List<Tuple2<String, Row>> list = new ArrayList<Tuple2<String, Row>>();
                for(int i = 0; i < 100; i++) {
                    list.add(new Tuple2<String, Row>(i + "_" + tuple._1, tuple._2));
                }
                return list;
            }

        });

// 將rdd1中分拆出來的致使傾斜的key的獨立rdd，每條數據都打上100之內的隨機前綴。
// 而後將這個rdd1中分拆出來的獨立rdd，與上面rdd2中分拆出來的獨立rdd，進行join。
JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD1 = skewedRDD.mapToPair(
        new PairFunction<Tuple2<Long,String>, String, String>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<String, String> call(Tuple2<Long, String> tuple)
                    throws Exception {
                Random random = new Random();
                int prefix = random.nextInt(100);
                return new Tuple2<String, String>(prefix + "_" + tuple._1, tuple._2);
            }
        })
        .join(skewedUserid2infoRDD)
        .mapToPair(new PairFunction<Tuple2<String,Tuple2<String,Row>>, Long, Tuple2<String, Row>>() {
                        private static final long serialVersionUID = 1L;
                        @Override
                        public Tuple2<Long, Tuple2<String, Row>> call(
                            Tuple2<String, Tuple2<String, Row>> tuple)
                            throws Exception {
                            long key = Long.valueOf(tuple._1.split("_")[1]);
                            return new Tuple2<Long, Tuple2<String, Row>>(key, tuple._2);
                        }
                    });

// 將rdd1中分拆出來的包含普通key的獨立rdd，直接與rdd2進行join。
JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD2 = commonRDD.join(rdd2);

// 將傾斜key join後的結果與普通key join後的結果，uinon起來。
// 就是最終的join結果。
JavaPairRDD<Long, Tuple2<String, Row>> joinedRDD = joinedRDD1.union(joinedRDD2);

解決方案七：使用隨機前綴和擴容RDD進行join

方案適用場景：若是在進行join操做時，RDD中有大量的key致使數據傾斜，那麼進行分拆key也沒什麼意義，此時就只能使用最後一種方案來解決問題了。

方案實現思路：

  • 該方案的實現思路基本和「解決方案六」相似，首先查看RDD/Hive表中的數據分佈狀況，找到那個形成數據傾斜的RDD/Hive表，好比有多個key都對應了超過1萬條數據。
  • 而後將該RDD的每條數據都打上一個n之內的隨機前綴。
  • 同時對另一個正常的RDD進行擴容，將每條數據都擴容成n條數據，擴容出來的每條數據都依次打上一個0~n的前綴。
  • 最後將兩個處理後的RDD進行join便可。

方案實現原理：將原先同樣的key經過附加隨機前綴變成不同的key，而後就能夠將這些處理後的「不一樣key」分散到多個task中去處理，而不是讓一個task處理大量的相同key。該方案與「解決方案六」的不一樣之處就在於，上一種方案是儘可能只對少數傾斜key對應的數據進行特殊處理，因爲處理過程須要擴容RDD，所以上一種方案擴容RDD後對內存的佔用並不大；而這一種方案是針對有大量傾斜key的狀況，無法將部分key拆分出來進行單獨處理，所以只能對整個RDD進行數據擴容，對內存資源要求很高。

方案優勢：對join類型的數據傾斜基本均可以處理，並且效果也相對比較顯著，性能提高效果很是不錯。

方案缺點：該方案更多的是緩解數據傾斜，而不是完全避免數據傾斜。並且須要對整個RDD進行擴容，對內存資源要求很高。

方案實踐經驗：曾經開發一個數據需求的時候，發現一個join致使了數據傾斜。優化以前，做業的執行時間大約是60分鐘左右；使用該方案優化以後，執行時間縮短到10分鐘左右，性能提高了6倍。

// 首先將其中一個key分佈相對較爲均勻的RDD膨脹100倍。
JavaPairRDD<String, Row> expandedRDD = rdd1.flatMapToPair(
        new PairFlatMapFunction<Tuple2<Long,Row>, String, Row>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Iterable<Tuple2<String, Row>> call(Tuple2<Long, Row> tuple)
                    throws Exception {
                List<Tuple2<String, Row>> list = new ArrayList<Tuple2<String, Row>>();
                for(int i = 0; i < 100; i++) {
                    list.add(new Tuple2<String, Row>(0 + "_" + tuple._1, tuple._2));
                }
                return list;
            }
        });

// 其次，將另外一個有數據傾斜key的RDD，每條數據都打上100之內的隨機前綴。
JavaPairRDD<String, String> mappedRDD = rdd2.mapToPair(
        new PairFunction<Tuple2<Long,String>, String, String>() {
            private static final long serialVersionUID = 1L;
            @Override
            public Tuple2<String, String> call(Tuple2<Long, String> tuple)
                    throws Exception {
                Random random = new Random();
                int prefix = random.nextInt(100);
                return new Tuple2<String, String>(prefix + "_" + tuple._1, tuple._2);
            }
        });

// 將兩個處理後的RDD進行join便可。
JavaPairRDD<String, Tuple2<String, Row>> joinedRDD = mappedRDD.join(expandedRDD);

解決方案八：多種方案組合使用

在實踐中發現，不少狀況下，若是隻是處理較爲簡單的數據傾斜場景，那麼使用上述方案中的某一種基本就能夠解決。可是若是要處理一個較爲複雜的數據傾斜場景，那麼可能須要將多種方案組合起來使用。好比說，咱們針對出現了多個數據傾斜環節的Spark做業，能夠先運用解決方案一和二，預處理一部分數據，並過濾一部分數據來緩解；其次能夠對某些shuffle操做提高並行度，優化其性能；最後還能夠針對不一樣的聚合或join操做，選擇一種方案來優化其性能。你們須要對這些方案的思路和原理都透徹理解以後，在實踐中根據各類不一樣的狀況，靈活運用多種方案，來解決本身的數據傾斜問題。

shuffle調優

調優概述

大多數Spark做業的性能主要就是消耗在了shuffle環節，由於該環節包含了大量的磁盤IO、序列化、網絡數據傳輸等操做。所以，若是要讓做業的性能更上一層樓，就有必要對shuffle過程進行調優。可是也必須提醒你們的是，影響一個Spark做業性能的因素，主要仍是代碼開發、資源參數以及數據傾斜，shuffle調優只能在整個Spark的性能調優中佔到一小部分而已。所以你們務必把握住調優的基本原則，千萬不要捨本逐末。下面咱們就給你們詳細講解shuffle的原理，以及相關參數的說明，同時給出各個參數的調優建議。

ShuffleManager發展概述

在Spark的源碼中，負責shuffle過程的執行、計算和處理的組件主要就是ShuffleManager，也即shuffle管理器。而隨着Spark的版本的發展，ShuffleManager也在不斷迭代，變得愈來愈先進。

在Spark 1.2之前，默認的shuffle計算引擎是HashShuffleManager。該ShuffleManager而HashShuffleManager有着一個很是嚴重的弊端，就是會產生大量的中間磁盤文件，進而由大量的磁盤IO操做影響了性能。

所以在Spark 1.2之後的版本中，默認的ShuffleManager改爲了SortShuffleManager。SortShuffleManager相較於HashShuffleManager來講，有了必定的改進。主要就在於，每一個Task在進行shuffle操做時，雖然也會產生較多的臨時磁盤文件，可是最後會將全部的臨時文件合併（merge）成一個磁盤文件，所以每一個Task就只有一個磁盤文件。在下一個stage的shuffle read task拉取本身的數據時，只要根據索引讀取每一個磁盤文件中的部分數據便可。

下面咱們詳細分析一下HashShuffleManager和SortShuffleManager的原理。

HashShuffleManager運行原理

未經優化的HashShuffleManager

下圖說明了未經優化的HashShuffleManager的原理。這裏咱們先明確一個假設前提：每一個Executor只有1個CPU core，也就是說，不管這個Executor上分配多少個task線程，同一時間都只能執行一個task線程。

咱們先從shuffle write開始提及。shuffle write階段，主要就是在一個stage結束計算以後，爲了下一個stage能夠執行shuffle類的算子（好比reduceByKey），而將每一個task處理的數據按key進行「分類」。所謂「分類」，就是對相同的key執行hash算法，從而將相同key都寫入同一個磁盤文件中，而每個磁盤文件都只屬於下游stage的一個task。在將數據寫入磁盤以前，會先將數據寫入內存緩衝中，當內存緩衝填滿以後，纔會溢寫到磁盤文件中去。

那麼每一個執行shuffle write的task，要爲下一個stage建立多少個磁盤文件呢？很簡單，下一個stage的task有多少個，當前stage的每一個task就要建立多少份磁盤文件。好比下一個stage總共有100個task，那麼當前stage的每一個task都要建立100份磁盤文件。若是當前stage有50個task，總共有10個Executor，每一個Executor執行5個Task，那麼每一個Executor上總共就要建立500個磁盤文件，全部Executor上會建立5000個磁盤文件。因而可知，未經優化的shuffle write操做所產生的磁盤文件的數量是極其驚人的。

接着咱們來講說shuffle read。shuffle read，一般就是一個stage剛開始時要作的事情。此時該stage的每個task就須要將上一個stage的計算結果中的全部相同key，從各個節點上經過網絡都拉取到本身所在的節點上，而後進行key的聚合或鏈接等操做。因爲shuffle write的過程當中，task給下游stage的每一個task都建立了一個磁盤文件，所以shuffle read的過程當中，每一個task只要從上游stage的全部task所在節點上，拉取屬於本身的那一個磁盤文件便可。

shuffle read的拉取過程是一邊拉取一邊進行聚合的。每一個shuffle read task都會有一個本身的buffer緩衝，每次都只能拉取與buffer緩衝相同大小的數據，而後經過內存中的一個Map進行聚合等操做。聚合完一批數據後，再拉取下一批數據，並放到buffer緩衝中進行聚合操做。以此類推，直到最後將全部數據到拉取完，並獲得最終的結果。

優化後的HashShuffleManager

下圖說明了優化後的HashShuffleManager的原理。這裏說的優化，是指咱們能夠設置一個參數，spark.shuffle.consolidateFiles。該參數默認值爲false，將其設置爲true便可開啓優化機制。一般來講，若是咱們使用HashShuffleManager，那麼都建議開啓這個選項。

開啓consolidate機制以後，在shuffle write過程當中，task就不是爲下游stage的每一個task建立一個磁盤文件了。此時會出現shuffleFileGroup的概念，每一個shuffleFileGroup會對應一批磁盤文件，磁盤文件的數量與下游stage的task數量是相同的。一個Executor上有多少個CPU core，就能夠並行執行多少個task。而第一批並行執行的每一個task都會建立一個shuffleFileGroup，並將數據寫入對應的磁盤文件內。

當Executor的CPU core執行完一批task，接着執行下一批task時，下一批task就會複用以前已有的shuffleFileGroup，包括其中的磁盤文件。也就是說，此時task會將數據寫入已有的磁盤文件中，而不會寫入新的磁盤文件中。所以，consolidate機制容許不一樣的task複用同一批磁盤文件，這樣就能夠有效將多個task的磁盤文件進行必定程度上的合併，從而大幅度減小磁盤文件的數量，進而提高shuffle write的性能。

假設第二個stage有100個task，第一個stage有50個task，總共仍是有10個Executor，每一個Executor執行5個task。那麼本來使用未經優化的HashShuffleManager時，每一個Executor會產生500個磁盤文件，全部Executor會產生5000個磁盤文件的。可是此時通過優化以後，每一個Executor建立的磁盤文件的數量的計算公式爲：CPU core的數量 * 下一個stage的task數量。也就是說，每一個Executor此時只會建立100個磁盤文件，全部Executor只會建立1000個磁盤文件。

SortShuffleManager運行原理

SortShuffleManager的運行機制主要分紅兩種，一種是普通運行機制，另外一種是bypass運行機制。當shuffle read task的數量小於等於spark.shuffle.sort.bypassMergeThreshold參數的值時（默認爲200），就會啓用bypass機制。

普通運行機制

下圖說明了普通的SortShuffleManager的原理。在該模式下，數據會先寫入一個內存數據結構中，此時根據不一樣的shuffle算子，可能選用不一樣的數據結構。若是是reduceByKey這種聚合類的shuffle算子，那麼會選用Map數據結構，一邊經過Map進行聚合，一邊寫入內存；若是是join這種普通的shuffle算子，那麼會選用Array數據結構，直接寫入內存。接着，每寫一條數據進入內存數據結構以後，就會判斷一下，是否達到了某個臨界閾值。若是達到臨界閾值的話，那麼就會嘗試將內存數據結構中的數據溢寫到磁盤，而後清空內存數據結構。

在溢寫到磁盤文件以前，會先根據key對內存數據結構中已有的數據進行排序。排序事後，會分批將數據寫入磁盤文件。默認的batch數量是10000條，也就是說，排序好的數據，會以每批1萬條數據的形式分批寫入磁盤文件。寫入磁盤文件是經過Java的BufferedOutputStream實現的。BufferedOutputStream是Java的緩衝輸出流，首先會將數據緩衝在內存中，當內存緩衝滿溢以後再一次寫入磁盤文件中，這樣能夠減小磁盤IO次數，提高性能。

一個task將全部數據寫入內存數據結構的過程當中，會發生屢次磁盤溢寫操做，也就會產生多個臨時文件。最後會將以前全部的臨時磁盤文件都進行合併，這就是merge過程，此時會將以前全部臨時磁盤文件中的數據讀取出來，而後依次寫入最終的磁盤文件之中。此外，因爲一個task就只對應一個磁盤文件，也就意味着該task爲下游stage的task準備的數據都在這一個文件中，所以還會單獨寫一份索引文件，其中標識了下游各個task的數據在文件中的start offset與end offset。

SortShuffleManager因爲有一個磁盤文件merge的過程，所以大大減小了文件數量。好比第一個stage有50個task，總共有10個Executor，每一個Executor執行5個task，而第二個stage有100個task。因爲每一個task最終只有一個磁盤文件，所以此時每一個Executor上只有5個磁盤文件，全部Executor只有50個磁盤文件。

bypass運行機制

下圖說明了bypass SortShuffleManager的原理。bypass運行機制的觸發條件以下：

  • shuffle map task數量小於spark.shuffle.sort.bypassMergeThreshold參數的值。
  • 不是聚合類的shuffle算子（好比reduceByKey）。

此時task會爲每一個下游task都建立一個臨時磁盤文件，並將數據按key進行hash而後根據key的hash值，將key寫入對應的磁盤文件之中。固然，寫入磁盤文件時也是先寫入內存緩衝，緩衝寫滿以後再溢寫到磁盤文件的。最後，一樣會將全部臨時磁盤文件都合併成一個磁盤文件，並建立一個單獨的索引文件。

該過程的磁盤寫機制其實跟未經優化的HashShuffleManager是如出一轍的，由於都要建立數量驚人的磁盤文件，只是在最後會作一個磁盤文件的合併而已。所以少許的最終磁盤文件，也讓該機制相對未經優化的HashShuffleManager來講，shuffle read的性能會更好。

而該機制與普通SortShuffleManager運行機制的不一樣在於：第一，磁盤寫機制不一樣；第二，不會進行排序。也就是說，啓用該機制的最大好處在於，shuffle write過程當中，不須要進行數據的排序操做，也就節省掉了這部分的性能開銷。

shuffle相關參數調優

如下是Shffule過程當中的一些主要參數，這裏詳細講解了各個參數的功能、默認值以及基於實踐經驗給出的調優建議。

spark.shuffle.file.buffer

  • 默認值：32k
  • 參數說明：該參數用於設置shuffle write task的BufferedOutputStream的buffer緩衝大小。將數據寫到磁盤文件以前，會先寫入buffer緩衝中，待緩衝寫滿以後，纔會溢寫到磁盤。
  • 調優建議：若是做業可用的內存資源較爲充足的話，能夠適當增長這個參數的大小（好比64k），從而減小shuffle write過程當中溢寫磁盤文件的次數，也就能夠減小磁盤IO次數，進而提高性能。在實踐中發現，合理調節該參數，性能會有1%~5%的提高。

spark.reducer.maxSizeInFlight

  • 默認值：48m
  • 參數說明：該參數用於設置shuffle read task的buffer緩衝大小，而這個buffer緩衝決定了每次可以拉取多少數據。
  • 調優建議：若是做業可用的內存資源較爲充足的話，能夠適當增長這個參數的大小（好比96m），從而減小拉取數據的次數，也就能夠減小網絡傳輸的次數，進而提高性能。在實踐中發現，合理調節該參數，性能會有1%~5%的提高。

spark.shuffle.io.maxRetries

  • 默認值：3
  • 參數說明：shuffle read task從shuffle write task所在節點拉取屬於本身的數據時，若是由於網絡異常致使拉取失敗，是會自動進行重試的。該參數就表明了能夠重試的最大次數。若是在指定次數以內拉取仍是沒有成功，就可能會致使做業執行失敗。
  • 調優建議：對於那些包含了特別耗時的shuffle操做的做業，建議增長重試最大次數（好比60次），以免因爲JVM的full gc或者網絡不穩定等因素致使的數據拉取失敗。在實踐中發現，對於針對超大數據量（數十億~上百億）的shuffle過程，調節該參數能夠大幅度提高穩定性。

spark.shuffle.io.retryWait

  • 默認值：5s
  • 參數說明：具體解釋同上，該參數表明了每次重試拉取數據的等待間隔，默認是5s。
  • 調優建議：建議加大間隔時長（好比60s），以增長shuffle操做的穩定性。

spark.shuffle.memoryFraction

  • 默認值：0.2
  • 參數說明：該參數表明了Executor內存中，分配給shuffle read task進行聚合操做的內存比例，默認是20%。
  • 調優建議：在資源參數調優中講解過這個參數。若是內存充足，並且不多使用持久化操做，建議調高這個比例，給shuffle read的聚合操做更多內存，以免因爲內存不足致使聚合過程當中頻繁讀寫磁盤。在實踐中發現，合理調節該參數能夠將性能提高10%左右。

spark.shuffle.manager

  • 默認值：sort
  • 參數說明：該參數用於設置ShuffleManager的類型。Spark 1.5之後，有三個可選項：hash、sort和tungsten-sort。HashShuffleManager是Spark 1.2之前的默認選項，可是Spark 1.2以及以後的版本默認都是SortShuffleManager了。tungsten-sort與sort相似，可是使用了tungsten計劃中的堆外內存管理機制，內存使用效率更高。
  • 調優建議：因爲SortShuffleManager默認會對數據進行排序，所以若是你的業務邏輯中須要該排序機制的話，則使用默認的SortShuffleManager就能夠；而若是你的業務邏輯不須要對數據進行排序，那麼建議參考後面的幾個參數調優，經過bypass機制或優化的HashShuffleManager來避免排序操做，同時提供較好的磁盤讀寫性能。這裏要注意的是，tungsten-sort要慎用，由於以前發現了一些相應的bug。

spark.shuffle.sort.bypassMergeThreshold

  • 默認值：200
  • 參數說明：當ShuffleManager爲SortShuffleManager時，若是shuffle read task的數量小於這個閾值（默認是200），則shuffle write過程當中不會進行排序操做，而是直接按照未經優化的HashShuffleManager的方式去寫數據，可是最後會將每一個task產生的全部臨時磁盤文件都合併成一個文件，並會建立單獨的索引文件。
  • 調優建議：當你使用SortShuffleManager時，若是的確不須要排序操做，那麼建議將這個參數調大一些，大於shuffle read task的數量。那麼此時就會自動啓用bypass機制，map-side就不會進行排序了，減小了排序的性能開銷。可是這種方式下，依然會產生大量的磁盤文件，所以shuffle write性能有待提升。

spark.shuffle.consolidateFiles

    • 默認值：false
    • 參數說明：若是使用HashShuffleManager，該參數有效。若是設置爲true，那麼就會開啓consolidate機制，會大幅度合併shuffle write的輸出文件，對於shuffle read task數量特別多的狀況下，這種方法能夠極大地減小磁盤IO開銷，提高性能。
    • 調優建議：若是的確不須要SortShuffleManager的排序機制，那麼除了使用bypass機制，還能夠嘗試將spark.shffle.manager參數手動指定爲hash，使用HashShuffleManager，同時開啓consolidate機制。在實踐中嘗試過，發現其性能比開啓了bypass機制的SortShuffleManager要高出10%~30%。

 

轉載自https://tech.meituan.com/spark-tuning-pro.html

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程