388
棱鏡-分布式實時計算的跟蹤校驗係統
該文章來自於阿裏巴巴技術協會（ATA）精選文章。
摘要：*目前，各種分布式實時計算係統已經在各大互聯網公司得到了廣泛應用。但是，這些實時係統的計算過程多不進行持久化，如果出現消息丟失等異常情況，通常很難定位問題出現的位置和具體原因，更無法做到主動發現消息丟失。對於廣告營銷等對消息準確性要求較高的業務場景來說,這種消息丟失的代價通常很高，即便很低的消息丟失率也會造成大量的財物損失。為此，阿裏媽媽開發了一套麵向分布式實時計算框架storm的實時跟蹤校驗係統——棱鏡係統，棱鏡係統實時記錄每條消息在storm上的處理路徑，主動發現消息丟失情況並報警。本文詳細介紹了幾位作者在開發棱鏡中遇到的困難和相應的解決方案，相信對其他分布式實時計算係統的跟蹤校驗係統也有一定的借鑒意義。*
項目wiki
1.介紹：
1.1.棱鏡項目的背景
如今的互聯網應用，日均數據處理量越來越大，對計算實時性的要求也越來越高。為此，各種分布式實時計算係統層出不窮，比較有名的有Yahoo S4,storm,puma等。其中storm由於具備易恢複、可擴展、高容錯等特性，目前被廣泛應用在阿裏媽媽營銷係統的各條業務線上。
不過，雖然storm的acker機製巧妙地保證了每條消息都會被完整處理，但這一機製也有其局限：1）必須正確的調用ack和fail接口，對於邏輯複雜的應用，誤調用在所難免；2）和storm相連接的外部更新係統不在acker的監控範圍內；3）如果消息處理出錯，無法確切知道在storm的哪個bolt出錯。
同時，由於Storm計算過程未進行持久化，無法查看曆史消息的處理路徑。這樣即便我們發現了某條消息有處理不完整的情況出現，也很難複現。
在棱鏡係統上線之前，阿裏媽媽的開發人員通常不能及時發現線上問題，有時甚至要晚於客戶發現。而為了定位這些問題，往往又需要分別查詢storm及storm訪問的多個外部分布式係統的日誌。由於這些日誌散落在集群的各台機器上，查找起來非常費時，且有些日誌受限於單機的存儲能力，保存的時效有限。這樣帶來的問題一是問題處理不及時，引發客戶投訴，客戶滿意度下降，造成財物損失；二是定位問題費時費力，定位成功率也不高；三是有些問題客戶自己可能也無法發現，或者發現後沒有及時反饋，導致bug一直在線上運行。
1.2.營銷實時係統對棱鏡的要求
為了解決1.1節中提到的問題，我們開發了棱鏡係統，它實時地監測著storm係統的運行情況，記錄下每條消息的處理路徑，方便開發人員複現問題。同時準實時地對每條消息的處理路徑進行校驗，在發現問題時主動向開發和運維人員報警，並提供了一鍵消息重發工具，在異常出現時做到及時恢複。
棱鏡係統在設計之初，就考慮到了營銷係統的特殊要求，提出了以下設計目標：
1. 精確性，作為一個跟蹤校驗係統，如果本身不夠精確，那麼必定引起使用者的不信賴，同時，營銷係統與收入、扣款直接相關的特性，也要求棱鏡必須做到足夠精確。
2. 實時性，實時係統的監測係統自然也需要滿足實時性，不然在用戶投訴後，再收到異常報警也沒有任何意義。
3. 對應用盡可能透明。阿裏媽媽的營銷係統有數十、上百個拓撲，如果升級十分困難的話，也必然會阻礙棱鏡的推廣。對於棱鏡，非常幸運的一點是，阿裏媽媽的多條業務線共用一套拓撲框架，不同的業務在這套框架下，通過不同的配置得到適用於自己的業務邏輯，故而棱鏡的改動可以集中在這個拓撲框架中，業務不需要關心。
2.棱鏡係統架構和跟蹤、校驗原理：
2.1.棱鏡係統總體架構：
![棱鏡係統的總體設計架構]
圖2.1 棱鏡係統的總體架構
圖2.1展示了棱鏡係統的總體架構。其中Spout/Bolt[A-C]是要監控的某個拓撲進行流式處理的多個分布式組件。Dispatcher和updated是storm訪問的外部更新係統，也在棱鏡的監控範圍內。ODPS（Open Data Processing Service）是阿裏巴巴研發的海量離線數據處理平台，提供了批量結構化數據的存儲和計算服務，棱鏡的跟蹤記錄、主動校驗計算、曆史數據查詢等都在ODPS平台上進行。從數據流的角度看，在storm等集群上收集到ODPS原始trace數據首先通過準實時merge將同屬於一條消息的所有trace聚集到一起，再通過消息校驗發現異常消息、觸發報警，詳細的數據流處理將在2.3節中介紹。
2.2.trace數據的打點方式
Spout/Bolt在接收和發射(emit)消息之後各打一條trace日誌，每條日誌以key-value形式記錄下SeqID（每條消息唯一的主鍵，在各個bolt間不變，用於對齊），execTime（在storm中打點時的時刻，精確到毫秒），appName（拓撲名），clusterName（集群名稱），boltName（spout或者bolt的名稱）， 機器ID以及一些其他自定義的個性化、多維度的信息。
舉例來講，對於如圖2.1中的拓撲結構所要處理的某條消息來說，它會在Spout/Bolt[A-C]的入口、出口各打一條trace日誌，共八條，這八條日誌組成該條消息的處理路徑。
2.3.消息跟蹤、主動檢驗流程
如2.2節所述，每一條消息在storm中進行處理時，都會產生多條散布在各台機器上的trace數據。 接下來需要對這些Trace數據進行收集、對齊和完整性校驗。整個流程包含以下幾個步驟：
1、 使用分布式係統日誌收集工具TimeTunnel將storm各節點上的trace日誌收集到ODPS上，表中每一行是一條原始的Trace日誌。
2、 對原始Trace數據進行解析、分列，並存儲於一張split表。這張表在查詢定位問題時使用，由於進行了分列，可以按照各種條件組成sql語句進行查詢。
3、 將Trace數據按照消息主鍵（SeqID）進行對齊，我們稱這個過程為merge流程。經過對齊，同一條消息的所有trace數據就被合並到一起，形成了一條消息的處理路徑（tracing path）。
4、 對每條消息的處理路徑進行完整性校驗（見2.4），如果發現有處理不完整的情況，則觸發報警。
以上步驟全部在ODPS以準實時的方式（小batch處理）進行處理，batch時間間隔可以進行調整。
2.4.主動校驗的原理
所謂主動校驗，就是檢查每條消息的處理路徑是否完整，是否走完了storm中預期要經過的所有bolt。同時，校驗時還必須考慮到以下幾種特殊情況
1、 可能有重傳。在消息處理出現問題時，storm會通過重發消息，防止瞬時故障造成的影響。
2、 可能有消息分裂。在storm中，一個bolt可以emit多個消息，我們稱這種情況為消息分裂（1分N）。
3、 可能有主動丟棄，這種情況下消息處理流程提前中止為正常情況。
為了解決上述三個問題，在棱鏡的主動檢驗中，采用了如下算法：
1、 對消息處理路徑上的trace進行排序，找到最後一次重傳的所有trace日誌，在主動校驗中僅考慮最後一次重傳。
2、 對最後一次重傳中每種spout/bolt入口、出口的trace日誌進行計數，查看是否有消息分裂的情況。如果有，則流程後方bolt的trace數也必須與之吻合。
3、 對於提前中止的情況，再檢查3.3節中將要介紹的ErrorCode，如果ErrorCode也表示異常，才認為該消息處理路徑真的出現異常。
2.5.Debug tracing的實現
利用2.1節中所述的架構，棱鏡還提供了debug tracing或者稱為即時tracing的功能。有些時候，開發同學需要在線上查看某條消息處理時的debug日誌（默認關閉）。這在以前隻能通過重啟拓撲，打開debug日誌選項，再到storm各節點上grep才能查看。
在棱鏡中，由於已經有了日誌收集，日誌分列的架構，通過添加一些截獲debug日誌輸出的代碼，就簡便地實現了這個功能。現在，開發同學隻要在Storm的輸入消息體裏加入debug=true的選項，棱鏡就會把這條消息的所有debug日誌都收集到split表中，供開發同學方便地查詢。
3.棱鏡開發遇到的困難和相應的解決方案
3.1. merge時間段區間劃分問題及兩階段merge的引入
圖3.1 merge時間段劃分問題
現在考慮2.3節中的merge流程。在棱鏡的處理中，為保持實時性，對這一聚合過程采取了分小batch處理的方式。例如，本次處理0分到5分的數據，下次處理5分到10分的數據。但是，這種時間區間劃分方式會導致某些同屬一條消息的處理路徑被切分到不同batch。圖3.1給出了這一問題的圖形化描述，其中，每個彩色長方形代表一條消息的處理路徑，長方形上沿對應最早的spout的execTime時間，下沿對應最後一條trace日誌的execTime時間。黑線代表批處理時間區間的某個邊緣（如下邊緣）的可能劃分位置，從圖中可以看出，不管這個時間邊緣劃在哪裏，都會導致某條消息處理路徑被一分為二。
為解決這一問題，我們將merge流程劃分為兩個階段，分別稱為merge1和merge2。在merge1階段，以15分鍾為一個時間區間進行一次聚合，這樣部分位於區間邊緣的消息路徑（tracing path）將不可避免地被切分到兩個區間。所以在merge1做消息聚合時，將本次處理區間內同一path的每條trace的execTime統一設置為該path所有trace中最小的execTime，這樣一來path在時間線上的表示也就由“方塊”壓縮成了“直線”，因此在merge2劃分時間區間時就不會再遇到時間區間邊緣的劃分問題。之後在merge2階段，以merge1時間區間的中點為邊緣再進行一次跨merge1分區的聚合操作，將merge1時被劃分到不同分區的同一消息路徑重新聚集到一起。
3.2.對重傳的處理和時間戳不準促使UUID的引入
在2.3節介紹的主動校驗算法中，需要對消息路徑上的trace進行排序，找出最後一次重傳的所有trace。理論上，這隻需要按每條trace的execTime大小排序就可以了。但是，作為一個分布式係統，storm各台機器間的時間戳會有最多5毫秒的誤差，這會打亂排序的順序。
或者，既然我們知道消息流過spout/bolt的順序，那按這個順序進行排序不是也可以嗎？這在正常情況下的確可以。不過重傳會導致這種排序算法失靈。
為此，我們為每一次重傳引入一個唯一的UUID，它在spout中通過函數System.nanotime()獲得，可以精確到納秒。這樣，在多次重傳間就可以利用execTime+UUID得到每次重傳發生的相對位置，在一次重傳內，再利用Bolt間的相對位置確定排序順序。
3.3.有合理丟棄情況存在，棱鏡引入ErrorCode
對於有些消息來說，沒有走完整套消息處理流程是正常情況。為了不產生誤報，我們在棱鏡中定義了一套返回值ErrorCode，用來表征處理流程中的消息丟棄是否符合預期。隻要相關業務線插件按照這套ErrorCode上報自己的處理結果，棱鏡就可以提供精準的主動校驗。
4.使用棱鏡的性能消耗
圖4.1列出了兩個拓撲使用棱鏡前後的性能對比。可以看出，棱鏡造成的時延增加在16%以內，QPS下降約在5%。
圖4.1 棱鏡的性能消耗
5.棱鏡的應用場景分析
目前，棱鏡項目已經在阿裏媽媽的搜索營銷、展示營銷、定向廣告等多個業務場景的多個storm拓撲上得到了廣泛的應用，取得了豐碩的成果。但是，棱鏡的應用場景絕不僅僅局限於storm，任何具有分布式、消息流式處理（實時或非實時的）架構的係統都可以通過接入類棱鏡的trace收集(timetunnel) + ODPS + 監控數據流pipeline的外部監控係統獲得曆史消息查詢和消息處理實時監測的好處，令這些係統由黑盒變白盒，為發現問題、定位問題提供方便。
6. 項目成果
棱鏡上線以後，取得了激動人心的效果。消息丟失的問題得到了全麵修複，日均丟消息數量從之前的280餘條降為現在的接近0條，相當於每年減少數百萬的財物損失。調查、定位問題的效率也大幅提升了20倍以上，例如淘寶直通車線上的某個問題，之前定位共花費了3人日，現在定位類似問題隻需要簡單執行一條命令就可以快速複現。由於有了主動監測報警，新出現的消息丟失可以得到快速發現和立即解決，相關投訴減少90%以上，提升了客戶滿意度。曆史消息處理路徑查詢也變得簡單方便，圖6.1和6.2展示了棱鏡提供的console和WebUI兩種查詢工具的打印效果。利用這一工具，業務同學可以簡便、快速地查詢最近20天內storm所有消息的處理路徑。
圖6.1 console工具的打印效果
圖6.2 webUi工具的打印效果
同時，利用這些沉澱下的數據，我們每天進行一次日消息處理量的全量統計，用郵件、網頁報表的形式將統計結果展示給各業務方，及時發現消息處理量的異常波動。
7. 總結
本文介紹了棱鏡這一分布式實時係統的跟蹤校驗係統，以及我們在開發中遇到的困難和相應解決方案。目前，棱鏡係統已經在阿裏媽媽內部得到了廣泛應用，使用方在不修改代碼的情況下就可以享受到棱鏡帶來的曆史消息快速查詢的好處。
棱鏡的上線，讓Storm由黑盒變成了白盒，為阿裏媽媽的開發人員及時發現和定位問題提供了極大的幫助。通過小batch的批處理方案，棱鏡做到了精確的準實時跟蹤校驗。據我們所知，這也是目前業界首創的解決方案。
所以，我們今天將棱鏡的架構和工程中的難點介紹出來，希望可以為其他分布式流處理係統的跟蹤校驗方案提供一些借鑒。
最後更新：2017-04-01 13:37:10