深度解析 Twitter Heron 大資料實時分析系統
阿新 • • 發佈：2019-02-06
所有的老的生產環境的topology已經執行在Heron上， 每天大概處理幾十T的資料， billions of訊息
（3）認為Storm設計不合理的地方
3.1 一個executor 存在2個執行緒， 一個執行執行緒， 一個傳送執行緒， 並且一個executor執行多個task， task的排程完全依賴來源的tuple， 很不方便確認哪個task出了問題。
3.2 因為多種task執行在一個worker中， 無法明確出每種task使用的資源， 也很難定位出問題的task，當出現效能問題或其他行為時， 常用就是重啟topology， 重啟後就好了，因為task進行了重新排程
3.3 日誌打到同一個檔案中，也很難查詢問題，尤其是當某個task瘋狂的列印日誌時
3.4 當一個task掛掉了，直接會幹掉worker，並強迫其他執行好的task被kill掉
3.5 最大的問題是，當topology某個部分出現問題時， 會影響到topology其他的環節
3.6 gc引起了大量的問題
3.7 一條訊息至少經過4個執行緒， 4個佇列， 這會觸發執行緒切換和佇列競爭問題
3.8 nimbus功能太多， 排程/監控/分發jar/metric report， 經常會成為系統的bottleneck
3.9 storm的worker沒有做到資源保留和資源隔離， 因此存在一個worker會影響到另外的worker。 而現有的isolation排程會帶來資源浪費問題。 Storm on Yarn也沒有完全解決這個問題。
3.10 zookeeper成為系統的瓶頸， 當叢集規模增大時。 有些系統為了降低zk心態，新增了tracker，但tracker增加了系統運維難度。
3.11 nimbus是系統單點
3.12 缺乏反壓機制
3.12.1 當receiver忙不過來時， sender就直接扔棄掉tuple，
3.12.2 如果關掉acker機制， 那無法量化drop掉的tuple
3.12.3 因為上游worker執行的計算就被扔棄掉。
3.12.4. 系統會變的難以預測(less predictable.)
3.13 常常出現效能問題， 導致tuple fail， tuple replay， 執行變慢
3.13.1 不良的replay， 任意一個tuple失敗了，都會導致整個tuple tree fail， 不良的設計時（比如不重要的tuple失敗），會導致tuple輕易被重發
3.13.2 當記憶體很大時，長時間的gc，導致處理延時，甚至被誤殺
3.13.3 佇列競爭
Heron設計原則：
Topology Manager
1. tm伴隨整個topology生命週期， 提供topology狀態的唯一contact （類似yarn的app master）
2. 可以一主多備， 大家搶佔zk 節點， 誰勝出，誰為master， 其他為standby
（1） 一個task 一個程序，
（2） 所有的程序之間通訊都是使用protocol buffer
（3） 一個gateway執行緒， 一個執行執行緒。 gateway執行緒負責和外圍通訊， sm/mm。 執行執行緒和現有storm的執行執行緒非常類似。執行執行緒會收集所有的metrics，然後傳送給gateway執行緒。
（4）這個data-in/data-out佇列會限定大小， 當data-in 佇列滿了的時候， gateway執行緒停止從local SM 讀取資料。同理如果data-out佇列滿，gateway會認為local SM不想接受更多的資料。 執行執行緒就不再emit或執行更多的tuple。
（5）data-in/data-out佇列大小不是固定， 如果是固定時， 當網路顛簸時，會導致記憶體中大量資料堆積無法傳送出去，並觸發GC, 並導致進一步的降級。因此是動態調整， 定期調整佇列大小。 如果佇列的capacity超過閥值時， 對其進行減半。這個操作持續進行指導佇列的capacity維持在一個穩定的水位或0。這種方式有利避免GC的影響。 當佇列的capcity小於某個閥值時， 會緩慢增長到配置大小或最大capacity值。
為什麼要重新設計Heron：
【題外話】這裡完全引用作者吐槽的問題， 不少問題，其實JStorm已經解決（1）debug-ability 很差， 出現問題，很難發現
1.1 多個task執行在一個系統程序中， 很難定位問題。需要一個清晰的邏輯計算單元到物理計算單元的關係（2）需要一種更高階的資源池管理系統
2.1 可以和其他程式設計框架共享資源， 說白了，就是類似yarn/mesos， 而在Twitter就是Aurora 2.2 更簡單的彈性擴容和縮容 叢集 2.3 因為不同task，對資源需求是不一樣的， 而storm會公平對待每個worker， 因此會存在worker浪費記憶體問題。當worker記憶體特別大時， 進行jstack或heap dump時，特別容易引起gc，導致被supervisor幹掉 2.4 經常為了避免效能故障，常常進行超量資源分配， 原本100個core，分配了200個
（3）認為Storm設計不合理的地方
3.1 一個executor 存在2個執行緒， 一個執行執行緒， 一個傳送執行緒， 並且一個executor執行多個task， task的排程完全依賴來源的tuple， 很不方便確認哪個task出了問題。
3.2 因為多種task執行在一個worker中， 無法明確出每種task使用的資源， 也很難定位出問題的task，當出現效能問題或其他行為時， 常用就是重啟topology， 重啟後就好了，因為task進行了重新排程
3.3 日誌打到同一個檔案中，也很難查詢問題，尤其是當某個task瘋狂的列印日誌時
3.4 當一個task掛掉了，直接會幹掉worker，並強迫其他執行好的task被kill掉
3.5 最大的問題是，當topology某個部分出現問題時， 會影響到topology其他的環節
3.6 gc引起了大量的問題
3.7 一條訊息至少經過4個執行緒， 4個佇列， 這會觸發執行緒切換和佇列競爭問題
3.8 nimbus功能太多， 排程/監控/分發jar/metric report， 經常會成為系統的bottleneck
3.9 storm的worker沒有做到資源保留和資源隔離， 因此存在一個worker會影響到另外的worker。 而現有的isolation排程會帶來資源浪費問題。 Storm on Yarn也沒有完全解決這個問題。
3.10 zookeeper成為系統的瓶頸， 當叢集規模增大時。 有些系統為了降低zk心態，新增了tracker，但tracker增加了系統運維難度。
3.11 nimbus是系統單點
3.12 缺乏反壓機制
3.12.1 當receiver忙不過來時， sender就直接扔棄掉tuple，
3.12.2 如果關掉acker機制， 那無法量化drop掉的tuple
3.12.3 因為上游worker執行的計算就被扔棄掉。
3.12.4. 系統會變的難以預測(less predictable.)
3.13 常常出現效能問題， 導致tuple fail， tuple replay， 執行變慢
3.13.1 不良的replay， 任意一個tuple失敗了，都會導致整個tuple tree fail， 不良的設計時（比如不重要的tuple失敗），會導致tuple輕易被重發
3.13.2 當記憶體很大時，長時間的gc，導致處理延時，甚至被誤殺
3.13.3 佇列競爭
Heron設計原則：
（1）相容老的storm api
（2）實現2種策略， At most once/At least once
架構：
排程器
Aurora是一個基於mesos的通用service scheduler， Hero基於Aurora 實現了一套Topology Scheduler， 並且這個排程器已經提供了一定的抽象，可以移植到yarn/mesos/ec2（我的理解應該稍加修改就可以執行在其他通用型排程器上） 2/ 第一個container 執行 Topology Manager（TM）， 其他的container 內部會執行一個Stream manager/Metrics Manager 和多個Heron Instance。 這裡一個container類似一個docker感念，表示一個資源集合，是Aurora的排程單元， 多個container可以執行在一臺機器上， 分配多少container由Aurora根據現有資源情況進行分配， 另外一個container設定了cgroup
Topology Manager
1. tm伴隨整個topology生命週期， 提供topology狀態的唯一contact （類似yarn的app master）
2. 可以一主多備， 大家搶佔zk 節點， 誰勝出，誰為master， 其他為standby
Stream manager（SM）
最大的改變就是源自Stream manager， Stream manager就相當於一個container的tuple的匯流排（hub）。 所有的Hero Instance（HI）都連線SM進行send/receive 如果container內部一個HI 傳送資料到另外一個HI，走的是本地快速通道。Backpressure 反壓機制
當下遊處理速度變慢後，通過反壓機制，可以通知上游進行減速， 避免資料因buffer被塞滿而丟失，並因此帶來資源浪費。TCP 反壓：
當一個HI 處理慢了後，則該HI的接收buffer會被填滿， 緊接著本地SM的sending buffer被填滿， ? 然後會傳播到其他的SM和上游HI。 這個機制很容易實現，但在實際使用中，存在很多問題。因為多個HI 共用SM， 不僅將上游的HI 降速了，也把下游的HI 降速。從而整個topology速度全部下架，並且長時間的降級。Spout 反壓。
這個機制是結合TCP 反壓機制， 一旦SM 發現一個或多個HI 速度變慢，立刻對本地spout進行降級， 停止從這些spout讀取資料。並且受影響的SM 會發送一個特殊的start backpressure message 給其他的sm，要求他們對spout進行本地降級。一旦出問題的HI 恢復速度後，本地的SM 會發送 stop backpressure message 解除降級。Stage-by-Stage 反壓
這個類似spout反壓，但是一級一級向上反壓。 Heron最後採用的是spout反壓， 因為實現比較簡單，而且降級響應非常迅速。 並且可以很快定位到那個HI 處理速度慢了。 每個socket channel都綁定了一個buffer， 當buffer 的 queue size超過警戒水位時，觸發反壓，減少時，接觸反壓。 這種機制，不會丟棄tuple，除了機器宕機。 topology可以設定開啟或關閉。Heron Instance
（1） 一個task 一個程序，
（2） 所有的程序之間通訊都是使用protocol buffer
（3） 一個gateway執行緒， 一個執行執行緒。 gateway執行緒負責和外圍通訊， sm/mm。 執行執行緒和現有storm的執行執行緒非常類似。執行執行緒會收集所有的metrics，然後傳送給gateway執行緒。
（4）這個data-in/data-out佇列會限定大小， 當data-in 佇列滿了的時候， gateway執行緒停止從local SM 讀取資料。同理如果data-out佇列滿，gateway會認為local SM不想接受更多的資料。 執行執行緒就不再emit或執行更多的tuple。
（5）data-in/data-out佇列大小不是固定， 如果是固定時， 當網路顛簸時，會導致記憶體中大量資料堆積無法傳送出去，並觸發GC, 並導致進一步的降級。因此是動態調整， 定期調整佇列大小。 如果佇列的capacity超過閥值時， 對其進行減半。這個操作持續進行指導佇列的capacity維持在一個穩定的水位或0。這種方式有利避免GC的影響。 當佇列的capcity小於某個閥值時， 會緩慢增長到配置大小或最大capacity值。