Flink大數(shù)據(jù)計(jì)算的機(jī)遇與挑戰(zhàn)

作者: 王紹翾（大沙）

本文來自于王紹翾在2018年08月11日Flink China Meetup。
王紹翾，花名“大沙”，加州大學(xué)圣迭戈分校計(jì)算機(jī)工程的博士，Apache Flink Commiter。目前在阿里負(fù)責(zé)Flink平臺(tái)以及生態(tài)的一些工作。

本文內(nèi)容如下：

流計(jì)算核心技術(shù)

Flink是德國data Artisans創(chuàng)造的，早期Flink主要是做偏批計(jì)算的，但是Spark在批處理上已經(jīng)有一定優(yōu)勢(shì)，正面競(jìng)爭(zhēng)沒什么意義，于是改變方向，基于chandy-lamport算法開始做流計(jì)算，完成后完美的解決了低延遲問題和狀態(tài)管理。

低延遲，快速容錯(cuò)

低延遲是Flink源生的，當(dāng)然保證了快速容錯(cuò)。大數(shù)據(jù)計(jì)算中job總是會(huì)失敗，所以需要能夠快速的恢復(fù)。如果平時(shí)延遲很低，但是job一失敗，恢復(fù)幾分鐘，肯定是無法接受的。

通用的API，易用性

Flink有了基礎(chǔ)的能力后，開始考慮通用的API，最開始的時(shí)候有了一些Java和Scala的一些API。但是發(fā)展到一定程度之后，因?yàn)锳PI不只是開放于開發(fā)，而是所有用戶。怎么樣更容易的滿足用戶的需求和支持用戶，這是流計(jì)算的很核心的一點(diǎn)。

彈性，高性能

彈性，高性能是大數(shù)據(jù)不變的主題。怎么樣確保引擎在上千臺(tái)機(jī)器不出問題的運(yùn)行，scalability很重要，包括Spark早期到一定規(guī)模遇到很多問題，當(dāng)然Blink已經(jīng)完美的解決了所有問題。在性能上，F(xiàn)link不僅是在流計(jì)算還是批處理上已經(jīng)有了絕對(duì)的優(yōu)勢(shì)。

流和批的統(tǒng)一

Flink的早期interface是非常弱的，包括Spark早期也是，于是流計(jì)算的社區(qū)開始討論流計(jì)算的SQL到底是什么樣子的，于是形成了兩派風(fēng)格，一派是認(rèn)為Streaming SQL是一種different SQL跟Batch Sql，另一派推的SQL跟Batch SQL是完全一致的。

為什么會(huì)說完全一致？流計(jì)算跟批計(jì)算一個(gè)基本的區(qū)別是，都是計(jì)算，但是流計(jì)算需要提前看到結(jié)果，這需要將結(jié)果提前發(fā)出，但是后面過來的數(shù)據(jù)會(huì)對(duì)前面的結(jié)果進(jìn)行修正，所以流計(jì)算跟批計(jì)算比較大的區(qū)別就是數(shù)據(jù)提前發(fā)出和數(shù)據(jù)修正，最終保證數(shù)據(jù)正確。

怎么來處理這個(gè)問題:

• 首先要告訴用戶API，怎么樣去計(jì)算完全是用戶的語義

• 另外兩點(diǎn)就是什么時(shí)候發(fā)出去，什么時(shí)候修正，這些跟SQL本身描述是沒什么關(guān)系的

• 所以傳統(tǒng)的ANSI SQL是完全可以描述流計(jì)算的，F(xiàn)link SQL的語義就是ANSI SQL

用戶要什么?

• 高性能

• 高級(jí)分析

• 容易開發(fā)

• 開箱即用

• 低延遲

cdn.xitu.io/2019/4/25/16a539a2cf310379?w=1055&h=563&f=jpeg&s=123260">

我們說的是大數(shù)據(jù)，而不僅僅是流計(jì)算。對(duì)于功能型的用戶，更關(guān)心的是易用性，如何做好分析，如何更好的開發(fā)，如何更容易上手。我沒學(xué)過計(jì)算機(jī)，但是學(xué)的是其他任何的一個(gè)行業(yè)可能是統(tǒng)計(jì)，生物，建筑，金融……，怎么樣才能更快的開發(fā)出來。

假如老板說，今天要部署Flink了，于是給了你50臺(tái)機(jī)器，到了第二天，你部署完畢了，作業(yè)跑起來了，老板嚇呆了，覺得你KPI非常的棒。所以開箱即用，更容易的去開發(fā)對(duì)用戶來說非常需要的。

傳統(tǒng)的批計(jì)算要追求performance，目前流計(jì)算對(duì)performance需求越來越大。

一.Flink的現(xiàn)狀和未來

知道了用戶想要的，我們看Flink現(xiàn)狀。

Flink目前被廣泛的用于超低延遲流計(jì)算場(chǎng)景中，但是Flink在批處理上其實(shí)已經(jīng)有非常高的處理性能，并且在API上流和批是統(tǒng)一的，在性能上和易用性上都有不錯(cuò)的表現(xiàn)。

帶著已知的事情和一點(diǎn)點(diǎn)未知的事情，來看看Flink能做的一些事情:流計(jì)算已經(jīng)非常成熟，批計(jì)算，AI的計(jì)算，包括TF ON Flink，training也好，prediction也好，任何計(jì)算。另外還有很大的一塊IOT，Hadoop Summit 中強(qiáng)調(diào)各種數(shù)據(jù)中，流的也好，批的也好，最終IOT的數(shù)據(jù)最大。雖然不是每個(gè)公司都會(huì)接觸IOT，但它絕對(duì)是一個(gè)很大的future。

1.阿里巴巴的Blink

Blink1.0實(shí)際上是enterprise版的Flink，主要專注與流計(jì)算上。

Blink2.0是一個(gè)統(tǒng)一的引擎，支持流處理和批處理，在其他方面，例如AI方面做了很大的改進(jìn)，在batch性能上已經(jīng)遠(yuǎn)超Spark。回饋社區(qū)也是這個(gè)版本。

2.Flink SQL Engine的架構(gòu)

我們先看一眼Flink SQL Engine，從上面開始有Query的API，有Query Optimization，下來會(huì)翻譯到DataSteam或者DataSet算子，然后Runtime，在各個(gè)集群上運(yùn)行。這個(gè)架構(gòu)在里面展開DataSteam和DataSet，可以看到幾個(gè)比較大的問題：

1. 在設(shè)計(jì)上，從來沒想過統(tǒng)一起來。最終Query Optimization翻譯完之后到DataStream或者DataSet是完全兩條獨(dú)立的pipline，而且往下的代碼是全完不復(fù)用的

2. 再一個(gè)可以看批計(jì)算，DataSet下面還有一個(gè)Optimized Plan，這兩層優(yōu)化給統(tǒng)一帶來很大的困難

3.Blink SQL Engine的架構(gòu)

我們把整個(gè)的SQL Engine換成上圖所示。從上層開始的API，到下面的Query Processor包括了Query Optimizer和Query Executor，當(dāng)做完這些發(fā)現(xiàn)，代碼大量的減少并被復(fù)用，一個(gè)job用同樣的SQL只需要標(biāo)識(shí)是Batch Mode還是Stream Mode，就會(huì)得到一樣的結(jié)果。

從API開始，翻譯成Logical Plan經(jīng)過Optimizer，再到類似寫DataStream的這種Physical Plan，我們可以看到在Optimizer之前的批跟流完全一樣，SQL一樣，Logical Plan也一樣。即用戶腦子里想的東西，在批和流中一模一樣。

二.優(yōu)化流計(jì)算的挑戰(zhàn)和機(jī)遇

在Optimizer之后，流和批有些不一樣。

批和流在一樣的地方就是一些簡(jiǎn)單的filter，predicate，projection還有joining reorder。

區(qū)別就是在流計(jì)算我們不去支持sort，因?yàn)槊織l數(shù)據(jù)一來，就要對(duì)之前的數(shù)據(jù)更新，就好比我讓在座的各位稱個(gè)體重，排個(gè)序，突然在座的哪位去上個(gè)廁所，體重變了，會(huì)影響很多人的排序，就需要改變大量的結(jié)果。所以在流上不去考慮類似sort的東西。但是流上因?yàn)橛衧tate的使用，怎么樣把它的性能變得很高，減少Retraction，怎么樣讓用戶的SLA用MicroBatch去優(yōu)化。

流計(jì)算上一旦變成SQL，就得跑標(biāo)準(zhǔn)的SQL測(cè)試，TPC-H，TPC-DS。我們看這個(gè)TPCH13，這個(gè)是測(cè)試的是用一張Customer表和一張Order表，需要做一次join和count。

這個(gè)計(jì)算在批計(jì)算上處理很方便，因?yàn)閮蓚€(gè)表就在那兒，它明顯的知道用戶表很小，它會(huì)把用戶表hash到各個(gè)地方先cache下來，然后讓訂單表流過去，這個(gè)性能非常高，因?yàn)镺rder這張最大的表只是不停的流而不落地。

在流計(jì)算上怎么處理呢？因?yàn)楦静恢罃?shù)據(jù)長(zhǎng)什么樣子，每邊一來就得存下來，左邊的Customer表來了之后存下來，因?yàn)橐恍兄恍璐嬉粋€(gè)，所以用的是ValueState，但是每個(gè)用戶有很多的Order，右邊的Order表則需要使用MapState，這個(gè)計(jì)算量非常大，性能非常差。怎么優(yōu)化呢，我們使用的SQL就有一個(gè)天然的好處Optimizer。SQL Engine有個(gè)rule就是轉(zhuǎn)移了上面的countAgg和下面的join，SQL里面有個(gè)代數(shù)優(yōu)化，先不考慮數(shù)據(jù)是什么樣子，我從代數(shù)上認(rèn)為中間這幅圖和最右邊這幅圖的計(jì)算結(jié)果是一致的，所以我可以先對(duì)兩邊進(jìn)行agg，我可以在Order那一邊先把每個(gè)用戶count完變成一行只有一個(gè)數(shù)據(jù)，預(yù)先處理好數(shù)據(jù)，這樣把Order表壓縮成和customer一樣大小的表，join上的開銷省了很多，state從龐大的MapState變成了輕量的ValueState，性能提升了25倍，這也是為什么SQL是有意義的。

對(duì)于一些流計(jì)算的特有優(yōu)化，比如知道用戶的SLA，有段時(shí)間就可以去配置mini-batch 。

做全網(wǎng)的count，那么用以上左圖的紅色和紫色，分別發(fā)送到一個(gè)地方去統(tǒng)計(jì)，不做預(yù)處理的話，紅色節(jié)點(diǎn)負(fù)載過高，很快就導(dǎo)致反壓。最好的辦法就是紅色和紫色的節(jié)點(diǎn)現(xiàn)在上游chain起來做預(yù)處理，相當(dāng)于把一個(gè)聚合分成兩部分，先做count，再做sum。

當(dāng)然上面的方案不總是有效，比如count distinct，它也需要按顏色group by還要按某一列去distinct，導(dǎo)致不同的數(shù)據(jù)無法被預(yù)聚合。所以在local-global上除了chain的方式還有shuffle的方式，相當(dāng)于shuffle兩次，也就是大家在流計(jì)算中所說的打散。第一次按distinct key去shuffle，第二次用group by的key去做shuffle。當(dāng)然這些都是SQL Engine都會(huì)自動(dòng)幫你做。

三.融入開源社區(qū)，參與開源開發(fā)

開源社區(qū)除了coding的貢獻(xiàn)外，還有文檔，生態(tài)，社區(qū)，產(chǎn)品，只要對(duì)這個(gè)開源的產(chǎn)品有幫助。更重要的是你在社區(qū)里面的活躍度，為社區(qū)解決什么問題。

作為一個(gè)用戶你可以提出一些問題，去mailing list回答問題，去做testing和report等等

作為一個(gè)開發(fā)你可以去review code，包括自己的idea，大的重構(gòu)。還可以幫助其他用戶回答問題。

Mailing lists：

<dev@flink.apache.org> 開發(fā)者提問交流。

<user@flink.apache.org> 用戶提問交流。

JIRA: https://issues.apache.org/jira/browse/FLINK

是社區(qū)的工作方式。Bug，feature，improvements提出的地方，每一個(gè)code的貢獻(xiàn)都會(huì)關(guān)聯(lián)到一個(gè)JIRA issue。

Wiki: https://cwiki.apache.org/confluence/display/FLINK

有許多文檔，包括大量FLIP，當(dāng)然也等著大家contribution。

那如何要參與開發(fā)呢？

1. 你要在社區(qū)提出自己的想法，收集一些建議。

2. 你還要了PMC，commiter對(duì)分別對(duì)哪部分code負(fù)責(zé)，你可以聯(lián)系他，讓他幫你review。

3. 可以依靠JIRA處理一些小的問題，但是比較重大的改進(jìn)還是需要依靠FLIP。

4. 完成之后，就需要去貢獻(xiàn)代碼，當(dāng)然要保證代碼的質(zhì)量，加入很多test case，當(dāng)你pull request時(shí)，會(huì)有很多人review你的代碼，沒有問題后就會(huì)merge上去。

更多資訊請(qǐng)?jiān)L問 Apache Flink 中文社區(qū)網(wǎng)站