争分夺秒：阿里实时大数据技术全力助战双11
摘要： 12月13-14日，由云棲社區與阿里巴巴技術協會共同主辦的《2017阿里巴巴雙11技術十二講》順利結束，集中為大家分享了2017雙11背后的黑科技。本文是《爭分奪秒：阿里實時大數據技術助戰雙 11》演講整理，主要講解了阿里巴巴實時大數據和機器學習技術，以及這些技術如何運用于阿里巴巴事業部，實現大數據升級，最終取得卓越的雙11戰果。
12月13-14日，由云棲社區與阿里巴巴技術協會共同主辦的《2017阿里巴巴雙11技術十二講》順利結束，集中為大家分享了2017雙11背后的黑科技。本文是《爭分奪秒：阿里實時大數據技術助戰雙11》演講整理，主要講解了阿里巴巴實時大數據和相關的機器學習技術，以及這些技術如何運用于阿里巴巴幾十個事業部，實現大數據升級，最終取得卓越的雙11戰果，內容如下。
分享嘉賓：
大沙，阿里巴巴高級技術專家，負責實時計算Flink SQL，之前在美國臉書任職，Apache Flink committer。
實時計算in阿里巴巴
1999年起，阿里從電商平臺開始不斷拓展業務，在金融、支付、物流、文娛各個領域衍生出眾多產品，例如依托于淘寶、天貓為主的電商平臺、阿里媽媽廣告平臺、螞蟻金服支付寶、阿里云、大文娛等。今天的阿里它已經不僅僅是一個電商平臺，而是一個龐大的應用生態。阿里巴巴目前是全球最大的電商平臺，擁有25個子公司，去年財年收入達到5500億美金。在阿里平臺上有近5億的用戶，相當于中國人口的1/3，每天有近1000萬用戶通過阿里平臺交易。
阿里儼然成為巨大的商業航母，在這艘航母上，在大量的用戶和應用之外，必然產生大量的數據。目前，阿里巴巴的數據量級已經達到EB級別，每天的增長量達到PB級別，每天實時計算數據也達到PB級，日常峰值處理的數據量可達到100GB/S，今年雙11更是達到了驚人的470GB/S。
實時計算在阿里巴巴內部應用廣泛。隨著新經濟體的發展，技術的革新和用戶需求的提升，人們越來越需要實時計算的能力，它的最大特征是數據是在變化的。接下來，舉兩個例子說明實時計算在阿里內部應用的場景：
1. 雙11大屏
每年雙11阿里都會聚合有價值的數據展現給媒體，GMV大屏是其中之一。整個GMV大屏是非常典型的實時計算，每條交易數據經過聚合展現在大屏之上。從DataBase寫入一條數據開始，到數據實時處理寫入HBase，最后展現在大屏之上，整個過程的鏈路十分長。整個應用存在著許多挑戰：
1) 大屏展現需要秒級延遲，這需要實時計算延遲在亞秒級別
2) 雙11大量數據需要在一個Job中聚合完成
3) Exactly-Once 保持數據計算的精確性
4) 系統高可用，不存在卡頓和不可用的情況
這個應用場景的SLA非常高，要求秒級延遲和數據的精確性，但它的計算并不復雜，接下來介紹更為復雜的應用。
2. 實時機器學習
機器學習一般有兩個重要的組件：Feature 和Model。傳統的機器學習對Feature的收集和Model的訓練頻率較低，無法適應不斷變化的應用需求。例如在雙11時，商品的價格、活動的規則與平時完全不同，依據之前的數據進行訓練得不到最優的效果。因此，只有實時收集Feature，訓練Model才能擬合出較為滿意的結果。為此，我們開發了這個平臺。
此實時機器學習平臺主要包括兩個部分：實時Feature計算和實時Model計算。這套系統同樣擁有很多挑戰，具體如下：
1) 機器學習需要采集各種各樣Metrics，存在許多DataSource
2) 維度多，如用戶維度、商品維度。維度的疊加甚至是笛卡兒積導致最后的Metrics是海量的，State非常巨大
3) 機器學習計算復雜，耗用大量CPU
4) 某些數據不能存在State中，需要外部存儲，存在大量外部IO
3. 實時A/B Testing
用戶的Query也有可能不停變化，典型的例子有實時的A/B Testing。
算法工程師在調優Model時會涉及多種Model，不同的Model有不同的計算模式和方法，產生不同的計算結果。因此，往往會有不同的Query訂閱實時數據，產生結果后根據用戶回饋迭代Model，最終得到最優模型。A/B Tesing的挑戰在于算法工程師往往計算很多Metrics，所有的Metrics都通過實時計算進行統計會浪費大量資源。
針對這個挑戰，我們設計了A/B Tesing的框架開發平臺。它用來同步算法工程師感興趣的Metrics進行聚合，收集起來并發送到Druid引擎。這樣，算法工程師根據不同Job篩選出結果的優劣，最后根據Druid對不同的Metrics進行統計分析，建立Model。
綜上，實時計算在阿里巴巴內部存在如下挑戰：
1) 業務龐大，場景多，導致邏輯復雜
2) 數據量大，擁有很多Job和機器
3) 低延遲，數據精確性，高吞吐量的需求
Flink的選定及優化
為了應對上述挑戰，我們調研了許多計算框架，最終選定Flink，原因如下：
1. Flink很好地引入和設計了State，基于State復雜的邏輯計算如join能得到很好的描述
2. Flink引入了Chandy-Lamport 算法，在此算法的支撐下可以完美實現Exactly-Once，并能在低延遲下實現高吞吐量。
然而，Flink在State、Chandy-Lamport 算法等方面還有很多缺陷，為此阿里開辟了名為Blink的項目。
Blink是開源Flink與阿里巴巴Improvement的結合，主要分兩大塊：
1. BlinkRuntime
不同公司在使用Flink時，存儲、調度和底層優化等方面會有不同，這一層不好與社區統一，我們稱之為BlinkRuntime。
原生的Flink只有比較底層的DataStream API，用戶在使用時需要設計大量的代碼，而且DataStream本身也有需要設計上的問題，每次修改都需要修改所有的用戶代碼。阿里巴巴團隊重新設計了流計算的Flink SQL并推回了社區。取名Flink SQL的原因，是因為我們希望和社區在API層保持統一，擁抱開源生態。
BlinkRuntime核心優化解密
1. 部署和模型的優化
優化包含以下幾點：
1) 解決大規模部署問題。Flink中一個Cluster只有一個JobMaster來管理所有的Job。隨著Job的不斷增加，單一的Master無法承接更多的Job，產生了瓶頸。因此，我們重構了架構，使每一個Job擁有自己的Master。
2) 早期的Flink中TaskManager管理很多Task，某一個Task的問題會導致TaskManager崩潰，進而影響其他Job。我們使每一個Job擁有自己的TaskManager，增強了Job的隔離。
3) 引入ResourceManager。ResourceManager可以和JobMaster通訊，實時動態地調整資源，達到最優的集群部署。
4) 我們不僅將這些優化應用在YarnCluster上，還應用到Mesos和Standalone的部署上。
有了這些工作，Flink就可以應用到大規模的集群部署
2. Incremental Checkpoint
Flink有不同的State存儲方式：內存和外部存儲。在面對多種State如機器學習時內存無法滿足存儲要求，這時往往需要外存。早期的Flink設計存在缺陷：checkpoint會把所有的data壓縮后，按照每一次checkpoint寫入磁盤。隨著State的不斷增大，checkpoint讀取和寫入的數據量十分巨大。這會導致Job的checkpoint無法在1分鐘內完成，這樣在failover時就會造成大量的回退，造成較長延遲。
因此，我們提出了Incremental Checkpoint。概括的說就是增量地進行checkpoint。由于歷史的checkpoint都已經完成，后面的checkpoint只需要將不同的數據放入存儲，這樣使checkpoint變得輕量，是的checkpoint可以在秒級完成，減小了failover的延遲。
3. 異步IO
很多時候我們將數據放在外部存儲，需要IO讀取數據。傳統的方式使用 Sync-IO，等待結果返回造成了較大延遲和CPU資源的浪費。為此，我們設計了Async-IO，允許異步地多線程地讀取數據。當數據到達時系統時，調用callback處理數據，需要保序時我們提供buffer暫時保存先到的數據，等前部數據全部到達后批量發送。系統的整體性能根據buffer大小實現幾十倍幾百倍的提升，這極大地提升了單機的CPU利用率和數據吞吐。
以上所述大部分優化已經推回社區。
Flink SQL核心功能解密
1. 阿里完成Apache Flink SQL 80%研發工作
目前，Apache Flink SQL 80%的功能是阿里巴巴貢獻的，包括兩百個提交和上十萬行代碼。使用Flink SQL的原因是因為我們發現了底層API給用戶的遷移、上線帶來的極大不便。那么，我們又為什么選擇SQL？原因如下：
1) SQL是描述性語言，SQL適合用來描述Job的需求。
2) SQL擁有比較好的優化框架，使用戶專注于業務邏輯而不用關心State等，使用門檻低。
3) SQL易懂，適合不同領域的人使用。
4) SQL的API十分穩定，更新Engine時不用更換用戶的Job。
5) 有些應用場景需要流式更新，批式驗證，一個SQL同時進行批計算和流計算能帶來巨大好處。批計算使用SQL，我們可以在此基礎上達到批和流的統一。
2. 流處理 VS 批處理
兩者的核心區別在于流處理的數據是無窮的而批處理的數據是有限的，這導致了其他三個區別：
1) 流處理不會結束并產生結果，批處理返回一個結果后結束。比方說，在雙11結束后，批處理計算當天所有買家花費的總金額，而流處理需要追蹤實時的交易金額，不停地計算。
2) 流計算需要做checkpoint并保留狀態，機器宕機時大量Job需要回滾。批計算則不需要，它的輸入數據往往是被持久化存儲過的。
3) 流數據會不斷更新，例如某一買家的花費總金額在不斷變化，而批處理的數據是一天花費的總金額，是固定的。流數據會被更改而批數據不會。
3. QueryConfiguration
為了定義什么時候產生流計算結果和怎么保留狀態，我們設計了Query Configuration，主要包括兩個部分：
1. Latency SLA
定義了從數據產生到展現的延遲，如雙11大屏是秒級別。
流數據中的State不能一直存在，用戶設置TTL（過期時間）來解決這個問題。
這樣，我們就消除了流和批的區別，實現統一。接下來我們需要考慮如何設計流式的SQL？
4. Dynamic-Table實現流式SQL
問題關鍵在于SQL在批處理中對表操作而流數據中并沒有表。因此，我們創建了數據會隨著時間變化的動態表。動態表是流的另一種表現形式，它們之間具有對偶性，即它們可以互相轉換而不破壞數據的一致性。以下是一個例子：
如圖，左邊是輸入流，我們為每一條數據產生Dynamic-Table，再將Table的變化用Changelog發送出去。隨著數據的輸入，兩邊的數據始終保持一致，這就證明了Dynamic-Table并沒有丟失語義和數據。
這樣，我們就可以根據表做SQL。我們將Stream理解為一個個Dynamic-Table，動態查詢產生新的Table。值得一提的是，Dynamic-Table是虛擬的一層，并不需要存儲落地。我們再來看一個例子：
如圖，當有輸入流的時候我們進行連續查詢。因為加入了連續查詢的convert，左右兩邊的流已經發生了變換。總之動態表大大支撐了我們在流上執行連續查詢SQL的能力。
5. 地球上不應該存在Stream SQL
通過例子我們發現有了Dynamic-Table不需要創造新的流式SQL，我們或許可以得出這樣的結論：地球上不應該有流式SQL。保持ANSI SQL是我們構建Flink SQL的原則，ANSI SQL完全可以描述Stream SQL。
6. ANSI SQL功能實現
此外，我們需要實現ANSI SQL的所有功能。阿里巴巴內部實現了所有batch框架所需要的功能：DML、DDL、QueryConf、UDF/UDTF/UDAF、連接join、撤回、Window聚合、查詢優化等等。現在詳細介紹其中幾項：
1) JOIN
流和動態表具有對偶性，一條SQL看似是Table的join，事實上是流的join。底層實現如下：
兩邊都來數據時立刻產出一個結果，例如order 5和6在接近的時間內到達。一邊數據先來會被存在State中并查詢對面的State，不存在則不輸出，直到對面數據來了之后產生結果。總之，兩個流具有兩個state，一邊的數據到達后存下來等待另外一邊數據，全部到達后inner join產生結果。另外，此圖還引入了流和外部表的join。機器學習時大量的數據存儲在HBase，連接HBase的操作實際上是在連接一個外部表，存在兩個模式：
a) Look up方式。流數據到達時查詢外部表得到結果。
b) 發送版本號給外部存儲service，然后存儲根據版本號給出結果。
值得一提的是，這個功能沒有新的設計和Query語法的引入（完全按照SQL-2011的標準實現的）。同樣，它在批計算上也適用。
2) Retraction
撤回是流計算的重要概念，舉一個例子作解釋：計算詞頻
英語文本到達后計算出每個單詞的頻次。Hello World Bark每個單詞出現一次，產生1——3的數據。當數據不斷更新增加一個Hello時，我們在詞頻表插入2——1的數據，但這樣就使頻次為1的單詞數出現了問題。出現問題的原因是因為流數據在不斷更新，這時就需要我們能檢測到這種錯誤并且擁有撤回機制。事實上，什么時候需要撤回可以使用SQL的Query Optimizer判斷，它是用戶無感知的。這就體現了SQL擁有天然優化框架的優勢。
如上圖，第一個場景不需要撤回而第二個需要，這完全是由優化框架決定而非用戶 。
3) Window聚合
Window聚合是Flink SQL的一個重要能力。這個例子中我們對每一個小時的數據聚合進行統計。我們還支持了滑動窗和Session Window。Window的聚合事實上是按照Window的標準做一個個小batch處理。
4) 查詢優化Query Optimization
除了添加新的功能，我們還做了大量的查詢優化。例如在Async-join服務表時，我們會自動優化成Async狀態的Table，改寫最終的Runtime實現。我們還對Multiple joins進行merge，做了micro-batching。如果沒有micro-batching，一條數據的到來就會伴隨著讀寫IO。有了micro-batching之后我們可以用兩次IO處理幾千條數據。另外還有join/aggregate pushdown和TopN的優化，現在舉例解釋TopN優化：
如上圖，我們想取銷售量前三的city，對用戶的Query有兩種解法：
a) 每有一條數據對保存的city進行排序，再截取前三個city，消耗大量存儲計算資源
b) Query Optimizer會自動識別查詢語句，只保存前面三個city，大大優化了計算和存儲復雜度
阿里巴巴實時計算應用
1. 阿里云流計算開發平臺
該平臺允許用戶編寫SQL，輸入數據產生輸出判斷邏輯正確與否。正確后用戶可以通過平臺在集群上部署，完成后檢測Job的運行情況。整個平臺完成了所有實時計算的需求，集開發、Debug、上線、部署、運維于一體，大大加速了用戶開發和上線的效率。值得一提的是，今年雙11期間大部分Job均通過這個平臺發布。阿里云，包括公共云、專有云也是通過這個平臺輸出給中小企業，讓他們分享阿里巴巴實時計算的能力。
2. 阿里實時機器學習平臺Porsche
本平臺是面向算法同學的UI拖拽平臺，提供標準組件供他們開發復雜組件。使用者將組件按照規則連接后可生成圖，圖在經過優化翻譯成SQL后可以上線和部署。本平臺免去了算法同學學習SQL的成本，主要對內開放。
雙11實時計算總結
上圖是阿里巴巴實時計算架構，底層是成千上百臺的機器，之上是統一部署的Resource Management和Storage，還有Blink Runtime和Flink SQL，用戶通過StreamCompute和Porsche平臺提交Job，阿里內部幾百個工程師已經提交了上千個Flink SQL Job。上述就是阿里巴巴實時計算的現狀。
在實時計算的助力下，雙11拿到1682億的戰果，實時計算的貢獻主要體現在以下幾點：
總結
以上是生活随笔為你收集整理的争分夺秒：阿里实时大数据技术全力助战双11的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 通过浪潮AIStation实现细粒度高性
- 下一篇： 写时复制就这么几行代码，还是不会？