spark 終止 運作_如何實作Spark過載保護

前言

因為我司将Spark大規模按Service模式使用，也就是Spark執行個體大多數是7*24小時服務的，然後接受各種ad-hoc查詢。通常最難受的就是被bad query 給拖死了，然後導緻服務不可用。那麼有沒有辦法讓Spark意識到危險時，及時止損，殺掉那個可能引起自己奔潰的query? 如果能做到那麼價值會很大。我們可以将将對應的query發給Spark執行個體的建構者以及對應的使用者，并且附帶上一些執行個體運作對應query的資訊，這樣可以有效的讓雙方溝通，優化查詢。

實作思路

肯定不能拍腦袋，畢竟這是一個複雜的事情，否則早就應該有非常成熟的工具出來了。我這裡也僅僅是最近兩天的思考，抛磚引玉，和大家一起探讨。

我拍腦袋的覺得，Spark挂掉常見的一般也就兩情況：

1. Spark Driver 沒有catch到的特定異常，然後導緻spark context關閉，最後停止正常服務。
2. Shuffle 導緻應用挂掉。

其中Shuffle導緻應用挂掉主要展現在：

1. Executor Full GC，driver以為Executor挂掉殺掉他，或者其他節點到這個節點拉資料，半天拉不動，然後connection reset。
2. Executor 真的就crash了，因為占用記憶體過大。
3. OOM,這個是shuffle申請記憶體時申請不到了，會發生，是以Spark自帶的OOM

然後因為超出Yarn記憶體限制的被殺，我們不做考慮。

其實Shuffle出現問題是Spark執行個體出現問題的主要原因。而導緻Shuffle出現問題的原因則非常多，最常見的是資料分布不均勻。對此，我們的監控思路也就有了：

1. 設定一個定時器，比如2s采集資料一次
2. 采集的資料大緻格式為 groupId, executorId, shuffleRead， shuffleWrites,其中 groupId為某一組job(比如MLSQL就是一個大腳本，每個腳本的任務都會有一個唯一的groupId)。采集的内容含義是，目前groupId涉及到的job已經長生的所有shuffle名額的快照。兩條資料相減，就是shuffle在某N個周期内的增量情況。shuffle包括bytes和records以及diskSpill三個次元。

首先我們考慮，一個Bad Query 對Spark 執行個體的危害性來源于對Executor的直接傷害。是以我們首先要計算的是每個Executor危險指數。

根據上面的資料，我麼可以計算Executor危害性的四個因子：

1. 計算shuffle 在在目前executor的非均衡程度，我們暫且稱之為非均衡指數。指數越高，情況越不妙。
2. shuffle速率，也就是每秒增長量。shuffle速率越低，則表示executor可能負載太高，出問題的機率就高。
3. shuffle bytes/ shuffle records 單記錄大小，單記錄越大，危險性越高。
4. 目前executor GC情況。機關時間GC時間越長，危險性越高。

現在我們得到一個公式：

危險指數 =a*非均衡指數 - b*shuffle速率 + c*單記錄大小 + d*gctime/persecond
           

因為本質上這幾個因子值互相是不可比的，直接相加肯定是有問題的。我們給了一個權重系數，同時我們希望這幾個因子盡可能可以歸一到(0-1)。具體優化方式如下：

1. 非均衡指數大機率可以歸到(0-1)
2. shuffle速率我們可以取一個取一個最大值（經驗），進而将其歸一到(0-1)
3. 單記錄大小我們也規定一個最大值。
4. gctime/persecond 肯定會是 (0,1)

是以最後的公式是：

某個job組對某個executor危險指數 = a*非均衡指數 
          - b*shuffle速率/最大速率 
          + c*單記錄大小/單記錄最大值 
          + d*gctime/persecond
           

其中 非均衡指數 = (shuffle r/w in executor - 平均 shuffle r/w) / (平均 shuffle r/w *

)

說明：

1. 當非均衡指數值為負數，則設定為0,當均衡指數大于1時，歸為1.

2. 

   為經驗值。也就是一個executor的shuffle負載小于平均值的多少倍時，我們認為還是能接受的。
3. 我們需要設定一個shuffle絕對資料量的門檻值，然後才對executor進行危險指數計算。譬如shuffle量大于1g,之後才開啟指數計算。

a,b,c,d 的值如何确定呢？因為在系統挂掉之前，我們的資料采集系統都會勤勤懇懇工作，找到這些讓系統挂掉的查詢，然後分别計算上面四個指數，然後得到一個最好的線性拟合即可。

上面是針對每個executor危險系數計算。實際上，整個叢集的安危取決于每一個executor是不是能扛過去。理論上A Executor扛不過去，B因為具有相同的資源配置，也會抗不過去，是以Bad Query最大的問題是會弄成連鎖反應，慢慢搞掉所有Executor. 是以我個人認為隻要有一個executor的危險指數過高，就應該終止Query。

同時，我們既可以監控全局的executor shuffle資料計算叢集危險指數，來确定叢集是不是有危險，一旦有危險，計算每個groupId的危險指數，然後殺掉topN危險指數最高的任務進而是叢集度過危險。分級計算可以保證我們計算的足夠快，同時也避免每個groupId的任務都是OK的，但是因為任務太多而導緻的問題。

額外

還有一個監控executor的變化情況，如果發現有N個executor短時被殺掉，那麼可以考慮終止目前所有query. 這可能會是一個簡單又會有效的方式。