第六章 大數據，6.2 雙11背後的大規模數據處理(做者：惠岸 朋春 謙樂)

6.2 雙11背後的大規模數據處理

1. 實時數據總線服務-TT

TimeTunnel（TT）在阿里巴巴集團內部是一個有着超過6年曆史的實時數據總線服務，它是前臺在線業務和後端異步數據處理之間的橋樑。從宏觀方面來看，開源界很是著名的Kafka+Flume的組合在必定程度上可以提供和TT相似的基礎功能；不一樣的是，在阿里巴巴的業務體量和訴求下，咱們有比較多的配置管控、資源調度、軌跡校驗和血緣識別等方面的工做。

TimeTunnel產品架構

 

1.1 Pub/Sub服務

經過上圖咱們清楚地看到，TT的核心部分是一個基於HBase作中間存儲的Pub/Sub服務，它提供了一個能支撐高讀寫比、大吞吐量和數據不丟的隊列服務。除此以外，基於平常運維考慮，咱們還支持了按時間seek和彈性伸縮的能力。

數據須要在Pub/Sub「落地」的需求一方面來自於業務上對熱點數據多份消費的考慮，另外一方面一些在線算法方面的應用須要常常性地對數據進行回放訓練，數據「落地」可以比較好地對先後臺進行解耦。事實上，TT裏最熱門的數據（例如天貓交易相關）有超過100倍的讀寫比；而從總體來看，僅雙11當天流出TT的數據也比流入的數據多了3倍以上。

選擇HBase做爲中間存儲的緣由是可以成本較低地複用基於HDFS的多副本存儲能力，以及HBase自身在提供讀寫服務時對於熱點數據的內存管理能力。圖 8是寫入TT的數據在HBase中的存儲模型，咱們在broker層面經過構造合理的rowkey來使得同一個分區下的數據可按rowkey順序scan；同時，由於在生成rowkey的時候咱們使用了broker上的時間戳做爲高位變量，所以能很方便地提供按時間seek的能力。

數據在HBase中的存儲模型

 

1.2數據採集

上圖左側黃色部分是TT的數據採集方案。咱們經過如下途徑來準實時地收集前臺業務產生的增量數據：

1. 依賴DRC實現對MySQL、OceanBase以及Oracle等前臺業務數據庫的增量變動進行捕捉解析；
2. 自研的日誌Agent部署在數十萬臺的應用服務器上，準實時地捕捉應用日誌的變化；
3. 和其餘一些內部主流存儲例如OTS進行打通；
4. 用戶採用TT提供的SDK主動寫入。

隨着集團內重要業務異地多活架構和全球化的發展，數據採集分散在跨越數千甚至上萬千米的多個IDC中；而與此相反，以Galaxy、ODPS爲表明的大數據計算服務則須要考慮充分地利用大集中的架構來提高吞吐能力。所以，不可避免地在數據採集過程當中須要對數據進行緩衝和壓縮以儘量下降長途鏈路對於吞吐量的負面影響。

矛盾的是，緩衝意味着前端產生的數據須要在採集端「等待」，也就意味着消費方看到的數據是延遲的。這對於像阿里媽媽這樣依賴TT作反做弊和實時計費的業務來說是很難接受的，數據延遲意味着資損，意味着用戶體驗的顯著降低。一樣地，壓縮也是要消耗採集端的服務器資源的，尤爲在雙11這樣的場景下，前臺業務對於採集端的功耗尤爲敏感。

遺憾的是，世界上歷來沒有一個只帶來好處而沒有任何弊端的事物，軟件和產品的設計中到處都是折衷和取捨。除了在技術層面將實現細節作到儘量極致，TT爲了服務這些不一樣的場景，也提供了一些可配置的參數例如buffersize、sendthreads、compressLevel等用來匹配用戶對延時、性能以及功耗的不一樣需求。

 

1.3 軌跡校驗

TT區別於其餘相似產品的最大之處，是咱們經過技術埋點實現了一套完整的數據軌跡校驗的方案——咱們稱之爲「門將」。軌跡校驗的目的在於經過監控的手段來保證「數據不丟」，設計得好，甚至能夠識別出數據的重複、亂序等狀況。

幾乎全部相似的產品都宣稱本身能作到「數據不丟」，固然也包括配備了「門將」以前的TT。有意思的是，幾乎全部相似的產品都會被「丟數據」這個問題困擾，一樣包括TT。由於我相信咱們必定有能力在軟件設計以及編碼實現方面作到「數據不丟」的承諾，但每每會在一些預期外的異常case、版本升級或者系統耦合的地方出現這樣那樣的紕漏，從而致使下游消費方看起來缺失了部分數據。

以日誌採集爲例，咱們碰到過由於操做系統的限制（請參閱max_user_watches相關的說明），inotify沒有通知到新文件的產生而發生整個文件漏採集；也碰到過由於軟件的bug在遞歸建立子目錄的狀況下出現了時序問題致使文件漏採集；還碰到過保存在應用服務器上的checkpoint文件被意外損壞致使的「丟數據」。這樣的案例實在太多，並且防不勝防。

因此，工業界真正的「數據不丟」我認爲是有完備的機制可以快速地發現數據丟失，考驗的是系統的監控能力。

上文提到過，TT支撐着阿里媽媽的實時反做弊和點擊計費業務；一樣地，螞蟻金服大量涉及資金覈對和商戶對帳的業務也將身家性命託付在TT上。這樣的業務不容許有任何緣由致使的數據正確性問題。

「門將」的核心思路是在採集端往TT寫入數據的同時，構造恰當的meta，將數據「鏈表化」，從而可以在「門將」的校驗服務裏對數據軌跡進行還原，進而和源頭進行校驗（圖 8）。

仍然以日誌採集爲例。在採集過程當中，咱們以ip+dev+inode+sign來惟一識別內網上的一個文件，在構造meta時記錄下當前數據包在原始文件中的offset和當前數據包的大小size，那麼對於同一個文件的多個數據包，經過offset和size就能快速地識別出文件內有沒有被重複採集或者遺漏採集。若是在恰當的時間內與這臺機器上ls命令獲得的結果進行比對，就很容易發現有沒有文件被漏採集。

 

1.4 小結

全部的技術實現都是業務需求的抽象，這些需求有可能來自於大多數用戶須要用到的功能，更有可能來自對上下游業務架構和場景的理解。數據總線服務是一個和業務架構耦合很是密切的基礎組件，阿里巴巴集團獨特的技術架構、多樣性的存儲方案和橫向平臺化的研發模式賦予了TT探究更復雜問題的原動力。

在2016年雙11這樣一個萬衆矚目的時間點，TT經過前期的軟件性能和機房規劃上的努力，高峯期單一集羣承擔了15GB/s的寫入和50GB/s的讀取流量，首次作到了對全部業務進行不降級服務。這對於咱們、對於搭建在TT上的衆多業務，都是極大的鼓舞。

 

2. Galaxy：大規模數據流處理技術

每一年雙11除了「折扣」，阿里人關注的另外一個焦點，就是面向全世界媒體直播的「實時大屏」（以下圖所示）。包括總成交量在內的各項指標，經過數字維度展示了雙11狂歡節這一是買家，賣家及物流小二共同創造的奇蹟！

圖：雙11媒體直播大屏

爲實現這一大屏，背後須要實時處理海量的、龐大電商系統各個模塊產生的交易日誌。例如雙11當天產生的日誌量達到了PB級別，而每秒處理的峯值更是高達近1億事件！

如此大規模、高吞吐和低延時計算，帶來一系列世界級的技術挑戰，包括：

1. 實時編程：流式的數據處理給業務邏輯的表達和推理帶來了不少的複雜性。特別面對不斷變化的業務需求，如何幫助用戶快速地編寫和驗證明時計算邏輯是相當重要的。
2. 低延時：實時計算強調計算延時和結果的時效性。例如實時大屏對計算延時特別敏感，每一年的雙11都超越前一年更早地達到相同的成交量，系統須要在秒級甚至毫秒級反應出每一筆交易。即便在流量高峯時（雙11晚0:00點）也須要保證延時！
3. 集羣利用率：爲提升資源利用率，咱們將不用業務的實時處理邏輯共享一個集羣。這樣的共享也帶來性能隔離的問題，即如何讓同一臺物理機上的不一樣邏輯任務不互相干擾。這也是大部分開源框架忽略的重要問題。
4. 嚴格容錯及數據一致性：隨着應對高吞吐而不斷擴大的集羣規模，各類軟硬件故障都難以免。如何保證明時計算在任何故障下都能產生準確、一致的計算結果，不遺漏、重複事件輸出，也不引發內部狀態的誤差，是另外一個重大挑戰。
5. 多樣化場景支持：隨着實時決策對業務的價值愈來愈多，系統還須要支持愈來愈複雜和多樣化的場景，如在線機器學習、結合圖計算實現的動態關係網絡分析等等。

下文介紹Galaxy的重要技術創新，簡要描述它們如何幫助應對以上技術挑戰。

 

2.1 SQL與增量計算——複用熟悉的離線思惟，自動實現增量（流式）計算

爲了簡化用戶編程，特別是利用原有的離線計算做業快速實現實時計算，Galaxy容許經過高層描述性語言，如用戶熟悉的SQL來編寫流計算做業。經過簡單幾行SQL代碼就能夠實現過濾、雙流關聯等業務邏輯。

在執行時，因爲數據是以流式進入系統的，用戶的SQL就像數據庫視圖同樣，被自動增量更新，並以必定的頻率輸出結果，供下游計算和展現。

這一獨特的編程設計，不只幫助用戶藉助熟悉的離線處理思惟表達實時計算邏輯，也由於一樣的程序能夠在離線系統運行，使得結果的對比變得易如反掌。

 

2.2 高性能優化引擎——實現低延時計算

用戶的SQL腳本通過編譯優化，生成數據流圖，而後運行於Galaxy的分佈式引擎之上。相比開源數據流引擎，Galaxy引擎在「阿里巴巴規模」下，面對真實複雜的業務場景作了不少優化。包括自適應的消息打包、自定義序列化、數據行+列壓縮、先進的內存管理、和內部緩存隊列和線程模型，以及基於下游向上遊「反向」傳遞壓力的流控策略等。

圖：Galaxy優化執行流和運行時模塊

通過以上一系列的優化，Galaxy相比去年提高了6倍左右的吞吐性能。下圖顯示了Galaxy相比開源系統的性能優點。在面對今年雙11 3倍於去年的峯值狀況下，表現很是穩健。

圖：開源框架性能對比，經過「窗口WordCount（6組參數）」基準測試獲取

 

2.3 靈活的資源調度

Galaxy面對阿里巴巴集團衆多業務場景，將不一樣業務放置於大規模（幾千臺服務器組成的）共享集羣中，以提升資源利用率。另外一方面也隨之帶來了「多租戶」環境下的做業資源隔離問題，它直接影響資源的有效利用和做業的計算性能。

通過多年的積累，Galaxy支持CPU、內存、網絡和磁盤I/O等多維度資源的隔離。例如，對於CPU的隔離支持靈活的min-max策略，既保證了每一個做業最基本的資源需求，也使的空閒的資源被最大限度利用。

圖：做業維度的CPU資源min-max共享模型

在此基礎上，Galaxy的資源調度還支持必定比例的「超賣」、做業優先級調度、動態負載均衡和微做業共享單一物理核等多種機制。對於資源消耗特別大的做業還支持動態按需分配（即資源的彈性分配）。在知足複雜的運維要求和實時計算連續性的同時，實現了高效的資源利用和性能隔離。

 

2.4 容錯與狀態管理

流計算須要連續處理可能無界的輸入和連續產生輸出。在長時間運行中，大規模計算集羣的各類軟件或硬件故障難以免。由此對於計算和中間結果（如內存狀態）的容錯就相當重要。爲了作到精確的容錯和故障恢復，保證結果的準確性。Galaxy支持多種靈活的容錯策略，以在不一樣計算特性下，權衡容錯資源消耗和恢復性能。如基於輸入的從新計算、狀態檢查點（checkpoint），甚至是多副本的狀態和計算容錯等。

特別是自動的分佈式增量檢查點功能，系統自動利用內存、本地磁盤和遠程存儲構成的多級存儲，在不影響流計算延時的狀況下異步實現了計算狀態的持久化。當有故障發生時，保存的狀態能夠被快速加載。這一切對用戶都是無感知的。

圖：自動利用多級存儲的流計算狀態管理

 

2.5 開放可編程API（兼容Apache Beam）

除了SQL這樣高層的描述語言和用戶自定義邏輯（UDF），Galaxy還支持Apache Beam API，以提供更爲靈活的實時邏輯編程。Beam是一個統一開放的大數據應用編程接口，能夠同時描述離線和在線邏輯，最先由Google提出。Beam提供了功能豐富的編程接口，能有效的處理有界、無界、亂序的數據流輸入。 下面顯示了經過Beam實現的流式WordCount的例子：

1.指定Runner(底層計算引擎)建立一個Pipeline。

2.使用Source在Pipeline上生成一個PCollection，輸入數據。

3.對PCollection應用Transforms操做，好比wordCount中的count操做。

4.對最後的PCollection應用Sink，輸出結果到外部存儲中。

5.Run Pipeline到底層的計算引擎中。

使用Beam實現WordCount代碼樣例

public static class CountWords extends PTransform<PCollection<String>,

    PCollection<KV<String, Long>>> {

  @Override

  public PCollection<KV<String, Long>> apply(PCollection<String> lines) {

    // Convert lines of text into individual words.

    PCollection<String> words = lines.apply(

        ParDo.of(new ExtractWordsFn()));

    // Count the number of times each word occurs.

    PCollection<KV<String, Long>> wordCounts =

        words.apply(Count.<String>perElement());

    return wordCounts;

  }

}

藉助Beam，用戶能夠利用高性能的Galaxy引擎，定製面向特定領域的系統交互接口。同時，Galaxy從此也將兼容更多生態（如Spark Streaming和Flink Streaming API）。

 

2.6 可視化集成開發平臺和自動化運維

Galaxy還提供了「一站式」的集成開發環境——貝葉斯（Bayes，https://data.aliyun.com/product/sc）和自動化運維平臺——特斯拉（Tesla）。經過它們，用戶能夠方便地管理流計算應用的生命週期，包括編程、調試、監控運維，極大地下降了流計算系統的使用門檻。

圖：貝葉斯集成開發環境

 

2.7 雙11的寶貴工程經驗！

爲保障系統在雙11平穩支撐業務，在以上功能基礎上，咱們還總結了完整的全鏈路保障方法：

• 主備雙鏈路容災：利用Galaxy對多副本執行的支持，面向雙11重點媒體大屏等實時業務，實現了跨機房的多鏈路副本。哪怕是整個機房的故障，都能在秒級自動切換到另外一副本上執行，保障了雙11系統高可用。
• 實時全鏈路監控：咱們從數據採集、讀取、消費、入庫各個環節都增長延時指標的埋點，能夠清晰地看到整條鏈路各個階段的延時，快速分析哪一個組件性能瓶頸。另外，針對做業自己運行狀況，好比輸入吞吐、流量、CPU和內存消耗，都作了實時分析和展現的系統，能在秒級發現做業的異常。
• 運維診斷工具：爲應對各類應急響應，咱們作了一套完整的運維診斷工具用於發現集羣熱點機器、熱點做業。在Tesla頁面上能快速找到集羣的熱點機器，經過「機器分析」工具查看這臺機器上實時跑的任務，而且能定位到相應的業務和用戶。經過「做業分析」工具能自動診斷異常，結合做業的優先級，實現了一鍵負載均衡、啓停、續跑等運維操做。

經過這些保障設施，雙11當天，即便在發生交換機硬件故障的狀況下，面向全球直播的媒體大屏業務並無受到任何影響！

 

2.8 小結

擁有這些和其它諸多能力，Galaxy已經具有了至關完善的實時計算能力，也提供了「一站式」的解決方案。今年雙11當天，Galaxy處理了PB級別數據，處理峯值達到了1億事件每秒，平均處理延遲在毫秒級！除了雙11媒體大屏，Galaxy還支撐着阿里巴巴集團內外衆多實時業務，包括數據運營、廣告營銷、搜索個性化、智能客服、物流調度、支付寶、聚划算等。

 

3. MaxCompute

每一年雙11都是阿里巴巴從最「前端」到最「後臺」全部系統整條鏈路的一次大考。電商在線系統的瀏覽和消費產生了大量數據，其數據量是日常的數倍到數十倍。這些數據最終要流到阿里巴巴的大數據計算服務—MaxCompute上來處理。

 

MaxCompute承載了阿里巴巴集團全部的離線計算任務，是集團內部核心大數據平臺。截止到目前支撐着每日百萬級規模的做業，整個系統擁有數萬臺機器，單集羣規模上萬，存儲已經到達了EB級別，天天有數千位活躍的工程師在平臺上作數據處理。

 

面對如此多的海量數據，首先須要可以低成本的將數據存儲下來。MaxCompute依託背後的飛天分佈式操做系統，將大量低成本PC服務管理起來。早在2013年，咱們基於對業務增加速度的判斷，發現系統的存儲立刻就要「撞牆」了，集羣的規模將要應付不了與日俱增的數據量。直到後來成立了5k項目組，對技術難點進行了攻堅，將單集羣規模擴大到了5000臺，阿里巴巴也成爲了中國首個獨立研發擁有大規模通用計算平臺技術的公司。

 

實際上單集羣規模到達上萬臺自己技術挑戰很是大，由於規模上來之後對系統設計要求很是高，整個架構不能有單點。可是整個業務規模決定了1萬臺機器是不夠的，所以MaxCompute抽象出來一個控制層，將分佈在各個不一樣數據中心的多個計算集羣統一管理，根據業務特色將不一樣的業務放在不一樣的計算集羣中，經過跨集羣複製，自動將數據在多個集羣中同步，使得用戶能夠把計算引擎當成一個平臺。

 

3.1 跨集羣複製和全局調度

運行在MaxCompute上的業務種類很是多，各個業務部門之間數據也有着錯綜複雜的依賴關係。若是剛好數據不在同一個地域/機房中，那麼就要進行數據的異地讀寫。好比分析支付寶的數據須要淘寶的數據，支付寶的數據和淘寶的數據並不在同一個機房集羣，那就須要跨集羣的去讀（直讀），或者將數據拷貝到本地再讀（跨集羣複製）。此外因爲數據是會被更新的，好比淘寶的數據更新了，這個時候要求支付寶的做業可以讀到最新版本的數據。生產任務有各自的基線時間，對處理時間有要求，不能因爲互訪數據致使任務延時太長。機房之間雖然有幾十到上百G的直連網絡專線，但其餘生產業務也對網絡帶寬有需求，互訪數據不能把帶寬都佔滿，須要有網絡流量控制。多個任務可能會訪問同一份異地數據，再考慮帶寬佔用的限制，因此訪問異地數據不能所有都經過直讀異地數據來解決，有的異地數據須要在本地複製一份以供屢次任務使用。

 

爲了解決這個問題，MaxCompute引入了跨集羣複製和全局調度機制。MaxCompute上全部的數據表和分區的元數據引入了版本號，當數據被更新時，其對應的計算集羣版本號也會更新。版本更新後，新版本所在的計算集羣的數據須要被複制到其餘計算集羣。但這個複製操做該什麼時候發生，須要考慮多種因素，好比任務完成時效要求，多集羣之間的帶寬大小等。對這些因素進行全局分析，才能利用動態預先調整，遠程讀，複製等多種手段作到全局調度。但這一全局分析須要系統運行數據才能進行。MaxCompute中的元數據、數據血緣關係的分析，以及整個系統運行過程當中產生的數據都會收集到元數據倉庫，這樣能夠利用平臺自己的數據分析能力來分析這些數據。這些數據被用來輔助MaxCompute平臺的工程師作數據化運營，甚至用來幫助系統自身進行優化。

 

3.2 基於歷史運行信息的優化

經過對天天運行的做業進行分析，咱們發現大部分做業都是重複執行的。這是數據倉庫中的一個典型的使用場景： 天天產生的新數據被同一套數據處理任務批量重複執行。這樣的場景帶來了巨大的優化機會。首先天天運行的任務所佔用的資源信息會被記錄下來，好比運行時佔用的CPU、內存和運行時間。工程師新開發的做業在第一次運行時，申請的CPU和內存通常都會和實際佔用的CPU、內存有所差異。若是申請的大於實際佔用的，會形成調度的時候爲做業多留資源，形成資源浪費，即資源的利用率降低。若是申請的小於實際佔用的，會形成一臺機器上調度的做業超過了機器可以承載的負荷。這兩種資源錯配的後果都會下降系統使用效率。最理想的結果是做業申請的資源與實際使用的可以徹底匹配。

 

HBO( History-ed Based Optimization) 基於歷史運行信息的優化就是經過收集做業的歷史運行記錄，根據實際CPU、內存佔用來指導做業合理設置的一種優化手段。它是對集羣資源分配的一種優化，歸納起來就是根據：任務執行歷史+集羣狀態信息+優化規則，獲得最優的做業資源配置。

 

HBO包含兩部分工做：

• 在線部分(Online)：查找是否存在相應的hbo優化計劃，若是有，則按照計劃進行資源分配並執行
• 離線部分(Offline)：從元數據倉庫和神農獲取任務的歷史執行記錄，按照必定的策略生成hbo優化計劃

 

下圖爲HBO的流程架構圖：

正常狀況下，這種基於歷史的優化效果很是顯著，由於做業整體數據量在天與天之間變化通常不會很大。但到了雙11，因爲當天產生的數據量一般是前幾天的數倍甚至數十倍，對於一些極限狀況須要作特殊處理。好比做業instance數會由於處理的數據量增大同步增加而超過單個做業instance數量上限。依託HBO的工做，能夠識別重複的做業、而且可以精準的對單個做業進行設置。利用這個能力，咱們能夠在節日前先對全部做業作一次分析，好比找出輸入表在去年雙11當天數據量顯著增漲的做業，或者找出instance數量已經快要接近極限的做業，將他們單個instance處理的數據量設大，順利度過雙11的考驗。以一樣的手法能夠指導製做針對雙11的預案，好比調整CPU、內存的設置、提早發現數據傾斜等等。