[[180070]]

阿里巴巴是世界上***的電子商務(wù)零售商。 我們?cè)?015年的年銷售額總計(jì)3940億美元，超過eBay和亞馬遜之和。阿里巴巴搜索(個(gè)性化搜索和推薦平臺(tái))是客戶的關(guān)鍵入口，并承載了大部分在線收入，因此搜索基礎(chǔ)架構(gòu)團(tuán)隊(duì)需要不斷探索新技術(shù)來改進(jìn)產(chǎn)品。

在電子商務(wù)網(wǎng)站應(yīng)用場(chǎng)景中，什么能造就一個(gè)強(qiáng)大的搜索引擎?答案就是盡可能的為每個(gè)用戶提供實(shí)時(shí)相關(guān)和準(zhǔn)確的結(jié)果。同樣一個(gè)不容忽視的問題就是阿里巴巴的規(guī)模，當(dāng)前很難找到能夠適合我們的技術(shù)。

Apache Flink®就是一種這樣的技術(shù)，阿里巴巴正在使用基于Flink的系統(tǒng)Blink來為搜索基礎(chǔ)架構(gòu)的關(guān)鍵模塊提供支持，最終為用戶提供相關(guān)和準(zhǔn)確的搜索結(jié)果。在這篇文章中，我將介紹Flink在阿里巴巴搜索中的應(yīng)用，并介紹我們選擇在搜索基礎(chǔ)架構(gòu)團(tuán)隊(duì)中使用Flink的原因。

我還將討論如何改進(jìn)Flink以滿足我們對(duì)Blink的獨(dú)特需求，以及我們?nèi)绾闻cdata Artisans和Flink社區(qū)合作，將這些更改貢獻(xiàn)給Flink社區(qū)。一旦成功地將我們的修改合并到開源項(xiàng)目中，我們會(huì)將現(xiàn)有系統(tǒng)從Blink轉(zhuǎn)移到Apache Flink。

Part 1： Flink在阿里巴巴搜索中的應(yīng)用

文檔創(chuàng)建

為用戶提供***搜索引擎的***步是創(chuàng)建可供搜索的文檔。在阿里巴巴的應(yīng)用場(chǎng)景中，文檔是由數(shù)百萬個(gè)商品列表和相關(guān)的商品數(shù)據(jù)組成。

因?yàn)樯唐窋?shù)據(jù)存儲(chǔ)在許多不同的地方，所以搜索文檔創(chuàng)建也是一個(gè)很大的挑戰(zhàn)，搜索基礎(chǔ)架構(gòu)團(tuán)隊(duì)將商品相關(guān)的所有信息匯總在一起并創(chuàng)建完整的搜索文檔。一般來說，整個(gè)過程分為3個(gè)階段：

1. 將不同來源(例如MySQL，分布式文件系統(tǒng))的所有商品數(shù)據(jù)同步到一個(gè)HBase集群中。
2. 使用業(yè)務(wù)邏輯將來自不同表的數(shù)據(jù)連接在一起，以創(chuàng)建最終的可搜索文檔。這是一個(gè)HBase表，我們稱之為'Result'表。
3. 將此HBase表導(dǎo)出為文件作為更新集合。 

這3個(gè)階段實(shí)際上是在經(jīng)典的“lambda架構(gòu)”中的2個(gè)不同的pipeline上運(yùn)行：全量構(gòu)建pipeline和增量構(gòu)建pipeline。

• 在全量構(gòu)建pipeline中，我們需要處理所有數(shù)據(jù)源，這通常是一個(gè)批處理作業(yè)。
• 在增量構(gòu)建pipeline中，我們需要處理在批處理作業(yè)完成后發(fā)生的更新。例如，賣家可能修改商品價(jià)格或商品描述以及庫存量的變化。這些信息需要盡可能快的反饋在搜索結(jié)果中。增量構(gòu)建pipeline通常是一個(gè)流式作業(yè)。

搜索算法實(shí)時(shí)A/B測(cè)試

我們的工程師會(huì)定期測(cè)試不同的搜索算法，并且需要盡可能快地評(píng)估出效果?，F(xiàn)在這種評(píng)估每天運(yùn)行一次，因?yàn)橄雽?shí)時(shí)分析效果，所以我們使用Blink構(gòu)建了一個(gè)實(shí)時(shí)A/B測(cè)試框架。

在線日志(展示，點(diǎn)擊，交易)由解析器和過濾器收集和處理，然后使用業(yè)務(wù)邏輯連接在一起。接下來聚合數(shù)據(jù)，將聚合結(jié)果推送到Druid，在Druid內(nèi)部，我們可以編寫查詢語句并對(duì)數(shù)據(jù)執(zhí)行復(fù)雜的OLAP分析，并查看不同算法的效果。 

在線機(jī)器學(xué)習(xí)

在這部分中Flink有兩個(gè)應(yīng)用場(chǎng)景。首先，我們來討論它在商品特征實(shí)時(shí)更新中的應(yīng)用。阿里巴巴搜索排序中使用的一些商品特征包括商品CTR，商品庫存和商品點(diǎn)擊總數(shù)。這些數(shù)據(jù)隨時(shí)間而變化，如果可以使用***的數(shù)據(jù)，我們就能為用戶提供更相關(guān)的搜索結(jié)果排序。Flink pipeline為我們提供在線特征更新，并大大提高了轉(zhuǎn)化率。

其次，每年的特定日子(如光棍節(jié))，有些商品折扣力度很大，有時(shí)甚至高達(dá)50%。因此，用戶行為也會(huì)發(fā)生很大的變化。交易量巨大，通常比我們?cè)谄綍r(shí)看到的高出很多倍。以前訓(xùn)練的模型在這個(gè)場(chǎng)景作用有限，因此我們使用日志和Flink流式作業(yè)構(gòu)建了在線機(jī)器學(xué)習(xí)模型，這個(gè)模型會(huì)將實(shí)時(shí)用戶行為數(shù)據(jù)反饋到系統(tǒng)中。結(jié)果在這些不常見但非常重要的營銷節(jié)日的轉(zhuǎn)換率有了很大的提升。 

Part 2： 選擇一個(gè)框架來解決問題

選擇Flink應(yīng)用到搜索基礎(chǔ)架構(gòu)中，我們?cè)谒膫€(gè)方面做過評(píng)估。 Flink在四個(gè)方面都滿足我們的要求。

• 敏捷： 我們期望能夠?yàn)檎麄€(gè)(2個(gè)pipeline)搜索基礎(chǔ)架構(gòu)流程維護(hù)一套代碼，因此需要一個(gè)高級(jí)的API來滿足我們的業(yè)務(wù)邏輯。
• 一致性： 賣方或商品數(shù)據(jù)庫發(fā)生的變化必須反饋在最終搜索結(jié)果中，因此搜索基礎(chǔ)架構(gòu)團(tuán)隊(duì)需要至少處理一次(at-least-once)的語義(對(duì)于公司中的一些其他Flink用例，要求正好一次(exactly-once)語義)。
• 低延遲： 當(dāng)商品庫存量發(fā)生變化時(shí)，必須盡快在搜索結(jié)果中得到體現(xiàn)。例如我們不想給售罄的商品給出很高的搜索排名。
• 成本： 阿里巴巴需要處理大量數(shù)據(jù)，以我們的集群規(guī)模，效率提高可以顯著的降低成本。因此我們需要一個(gè)高性能、高吞吐量的框架。

一般來說，有兩種方法來將批處理和流式處理統(tǒng)一起來。一種方法是將批處理作為基本出發(fā)點(diǎn)，在批處理框架上支持流式處理。這可能不符合真正意義上低延遲，因?yàn)橛梦⑴刻幚?micro-batching)模擬流式處理需要一些固定的開銷。因此，當(dāng)試圖減少延遲時(shí)，開銷的比例也會(huì)相應(yīng)增加。在我們的規(guī)模上，為每個(gè)微批量處理器調(diào)度1000個(gè)任務(wù)，需要重新建立連接并重新加載狀態(tài)。因此在某種程度上，微批處理方法代價(jià)太高將變得沒有意義。

Flink從另一個(gè)角度來解決這個(gè)問題，即將流式處理作為基本出發(fā)點(diǎn)，在流式處理框架上支持批量處理，將批處理作為流式處理的一種特殊情況。使用這種方法，不會(huì)丟掉我們?cè)谂幚砟Ｊ?批處理模式下流是有限的)下做出的優(yōu)化，你仍然可以做一些批量處理上的優(yōu)化。

Part 3： Blink是什么?

Blink是Flink的一個(gè)分支版本，我們做了一定的改進(jìn)以滿足阿里巴巴的一些特定需求。因此，Blink在幾個(gè)不同的集群上運(yùn)行，每個(gè)集群有大約1000臺(tái)機(jī)器，大規(guī)模集群的性能對(duì)我們來說非常重要。

Blink的改進(jìn)主要包括兩個(gè)方面：

Table API更完整，因此我們可以使用相同的SQL進(jìn)行批處理和流式處理。

更強(qiáng)大的YARN模式，但仍然100%兼容Flink的API和更廣泛的生態(tài)系統(tǒng)。

Table API

我們首先添加了對(duì)用戶自定義函數(shù)UDF的支持，方便在Flink中實(shí)現(xiàn)獨(dú)特的業(yè)務(wù)邏輯。我們還添加了一個(gè)流對(duì)流的join的功能，由于Flink對(duì)于狀態(tài)比較好的支持，所以實(shí)現(xiàn)起來比較容易。我們添加了幾個(gè)聚合函數(shù)以及滑動(dòng)窗口的支持，最有趣的一個(gè)是distinct_count。

(編輯注：FLIP-11涵蓋了與上述功能相關(guān)的一系列Table API和SQL改進(jìn)，對(duì)該主題感興趣的可以閱讀) 

接下來，我們將介紹運(yùn)行時(shí)改進(jìn)，可以分為四個(gè)不同的類別。

Yarn上的Blink

當(dāng)我們開始項(xiàng)目時(shí)，F(xiàn)link支持2種集群模式：standalone模式和YARN上的Flink。在YARN模式中，作業(yè)不能動(dòng)態(tài)請(qǐng)求和釋放資源，而是需要預(yù)先分配所有需要的資源。不同的作業(yè)可能共享相同的JVM進(jìn)程，這有利于資源利用和資源隔離。

Blink中每個(gè)作業(yè)都有自己的JobMaster，以根據(jù)作業(yè)需要請(qǐng)求和釋放資源。并且不同的作業(yè)不能在同一個(gè)Java進(jìn)程中運(yùn)行，這將在作業(yè)和任務(wù)之間得到***隔離。阿里巴巴團(tuán)隊(duì)目前正在與Flink社區(qū)合作，將這項(xiàng)工作貢獻(xiàn)給開放源代碼，改進(jìn)工作在FLIP-6(除了YARN之外還擴(kuò)展到其他集群管理器)中得到了體現(xiàn)。 

Operator縮放

在生產(chǎn)環(huán)境中，我們的客戶端可能需要改變Operator的并行性，但同時(shí)他們不想失去當(dāng)前狀態(tài)。當(dāng)我們開始使用Blink時(shí)，F(xiàn)link不支持在保持狀態(tài)的同時(shí)改變Operator的并行性。Blink引入了“bucket”的概念作為狀態(tài)管理的基本單位。有比任務(wù)更多的bucket，并且每個(gè)任務(wù)將被分配多個(gè)bucket。當(dāng)并行性改變時(shí)，我們將重新分配任務(wù)的bucket。 使用這種方法，可以改變Operator的并行性并維持狀態(tài)。

(編者注：Flink社區(qū)同時(shí)在Flink 1.2版本中解決了的這個(gè)問題 - 該功能在***版本的主分支中可用。Flink的“key groups”概念在很大程度上等同于上面提到的“bucket”，但是實(shí)現(xiàn)時(shí)使用的數(shù)據(jù)結(jié)構(gòu)略有不同。更多信息，請(qǐng)?jiān)贘ira查看FLIR-3755) 

增量Checkpoint

在Flink中，Checkpoint操作分為兩個(gè)階段：在本地獲取狀態(tài)快照，然后將狀態(tài)快照保存到HDFS(或另一個(gè)存儲(chǔ)系統(tǒng))，并且每個(gè)快照的整個(gè)狀態(tài)存儲(chǔ)在HDFS中。我們的狀態(tài)數(shù)據(jù)太大了，這種方法是不可行的，所以Blink只存儲(chǔ)修改的狀態(tài)在HDFS中，這能夠大大提高Checkpoint的效率。這種修改使我們能夠在生產(chǎn)環(huán)境中使用很大的狀態(tài)數(shù)據(jù)。 

異步I/O

我們的job在生產(chǎn)環(huán)境中很大瓶頸在訪問外部存儲(chǔ)器上，如HBase。為了解決這個(gè)問題，我們引入了異步I/O，我們將致力于為社區(qū)做出貢獻(xiàn)，并在FLIP-12中有詳細(xì)描述。

(編輯筆記：data Artisans認(rèn)為FLIP-12足夠強(qiáng)大，可以在不久的將來在某個(gè)時(shí)間擁有自己的獨(dú)立寫入。所以我們只是簡(jiǎn)單地介紹一下這里的想法，如果你想了解更多，可以查看FLIP writeup) 

Part 4： 阿里巴巴的Flink未來計(jì)劃是什么?

我們將繼續(xù)優(yōu)化我們的流式j(luò)ob，特別是更好地處理臨時(shí)傾斜(temporary skew)和慢節(jié)點(diǎn)(slow machines)，同時(shí)不會(huì)對(duì)反壓機(jī)制(backpressure)和故障快速恢復(fù)造成影響。正如在Flink Forward大會(huì)上大家討論的，我們認(rèn)為Flink作為批處理框架以及流式處理框架有著巨大潛力。我們正在努力利用Flink的批處理能力，希望在幾個(gè)月內(nèi)在生產(chǎn)環(huán)境中使用Flink批處理模式。

會(huì)議的另一個(gè)熱門話題是流式SQL，我們將繼續(xù)在Flink中添加更多的SQL支持和Table API的支持。阿里巴巴的業(yè)務(wù)持續(xù)增長，這意味著我們的job會(huì)越來越大，確保我們可以擴(kuò)展到更大的集群變得越來越重要。

非常重要的是，我們期待與社區(qū)繼續(xù)合作，以便將我們的工作貢獻(xiàn)回開源社區(qū)，以便所有Flink用戶都能從我們加入Blink的工作中受益。我們期待著在2017年Flink Forward大會(huì)上向您介紹我們的進(jìn)展情況。