摘要：四阿里相關(guān)工作介紹在阿里的成長(zhǎng)路線阿里是從年開(kāi)始調(diào)研年月啟動(dòng)項(xiàng)目，并完善在大規(guī)模生產(chǎn)下的一些優(yōu)化和改進(jìn)。年月已成為阿里的實(shí)時(shí)計(jì)算引擎。阿里在狀態(tài)管理和容錯(cuò)相關(guān)的工作正在做的工作，基于重構(gòu)方面的一些優(yōu)化，阿里也正在將功能做完善。

作者: 施曉罡

本文來(lái)自2018年8月11日在北京舉行的 Flink Meetup會(huì)議，分享來(lái)自于施曉罡，目前在阿里大數(shù)據(jù)團(tuán)隊(duì)部從事Blink方面的研發(fā)，現(xiàn)在主要負(fù)責(zé)Blink狀態(tài)管理和容錯(cuò)相關(guān)技術(shù)的研發(fā)

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

有狀態(tài)的流數(shù)據(jù)處理；

Flink中的狀態(tài)接口；

狀態(tài)管理和容錯(cuò)機(jī)制實(shí)現(xiàn)；

阿里相關(guān)工作介紹；

計(jì)算任務(wù)的結(jié)果不僅僅依賴于輸入，還依賴于它的當(dāng)前狀態(tài)，其實(shí)大多數(shù)的計(jì)算都是有狀態(tài)的計(jì)算。 比如wordcount,給一些word,其計(jì)算它的count,這是一個(gè)很常見(jiàn)的業(yè)務(wù)場(chǎng)景。count做為輸出，在計(jì)算的過(guò)程中要不斷的把輸入累加到count上去，那么count就是一個(gè)state。

狀態(tài)數(shù)據(jù)的存儲(chǔ)和訪問(wèn)；

狀態(tài)數(shù)據(jù)的備份和恢復(fù)；

狀態(tài)數(shù)據(jù)的劃分和動(dòng)態(tài)擴(kuò)容；

在傳統(tǒng)的批處理中，數(shù)據(jù)是劃分為塊分片去完成的，然后每一個(gè)Task去處理一個(gè)分片。當(dāng)分片執(zhí)行完成后，把輸出聚合起來(lái)就是最終的結(jié)果。在這個(gè)過(guò)程當(dāng)中，對(duì)于state的需求還是比較小的。

對(duì)于流計(jì)算而言，對(duì)State有非常高的要求，因?yàn)樵诹飨到y(tǒng)中輸入是一個(gè)無(wú)限制的流，會(huì)運(yùn)行很長(zhǎng)一段時(shí)間，甚至運(yùn)行幾天或者幾個(gè)月都不會(huì)停機(jī)。在這個(gè)過(guò)程當(dāng)中，就需要將狀態(tài)數(shù)據(jù)很好的管理起來(lái)。很不幸的是，在傳統(tǒng)的流計(jì)算系統(tǒng)中，對(duì)狀態(tài)管理支持并不是很完善。比如storm,沒(méi)有任何程序狀態(tài)的支持，一種可選的方案是storm+hbase這樣的方式去實(shí)現(xiàn)，把這狀態(tài)數(shù)據(jù)存放在Hbase中，計(jì)算的時(shí)候再次從Hbase讀取狀態(tài)數(shù)據(jù)，做更新在寫(xiě)入進(jìn)去。這樣就會(huì)有如下幾個(gè)問(wèn)題

流計(jì)算系統(tǒng)的任務(wù)和Hbase的數(shù)據(jù)存儲(chǔ)有可能不在同一臺(tái)機(jī)器上，導(dǎo)致性能會(huì)很差。這樣經(jīng)常會(huì)做遠(yuǎn)端的訪問(wèn)，走網(wǎng)絡(luò)和存儲(chǔ)；

備份和恢復(fù)是比較困難，因?yàn)镠base是沒(méi)有回滾的，要做到Exactly onces很困難。在分布式環(huán)境下，如果程序出現(xiàn)故障，只能重啟Storm，那么Hbase的數(shù)據(jù)也就無(wú)法回滾到之前的狀態(tài)。比如廣告計(jì)費(fèi)的這種場(chǎng)景，Storm+Hbase是是行不通的，出現(xiàn)的問(wèn)題是錢(qián)可能就會(huì)多算，解決以上的辦法是Storm+mysql，通過(guò)mysql的回滾解決一致性的問(wèn)題。但是架構(gòu)會(huì)變得非常復(fù)雜。性能也會(huì)很差，要commit確保數(shù)據(jù)的一致性。

對(duì)于storm而言狀態(tài)數(shù)據(jù)的劃分和動(dòng)態(tài)擴(kuò)容也是非常難做，一個(gè)很?chē)?yán)重的問(wèn)題是所有用戶都會(huì)在strom上重復(fù)的做這些工作，比如搜索，廣告都要在做一遍，由此限制了部門(mén)的業(yè)務(wù)發(fā)展。

Flink在最早設(shè)計(jì)的時(shí)候就意識(shí)到了這個(gè)問(wèn)題，并提供了豐富的狀態(tài)訪問(wèn)和容錯(cuò)機(jī)制。如下圖所示：

Keyed States

Operator States

2.1.1.Keyed States

Keyed States的使用

Flink也提供了Keyed States多種數(shù)據(jù)結(jié)構(gòu)類型

Keyed States的動(dòng)態(tài)擴(kuò)容

2.1.2.Operator State

Operator States的使用

Operator States的數(shù)據(jù)結(jié)構(gòu)不像Keyed States豐富，現(xiàn)在只支持List

Operator States多種擴(kuò)展方式

Operator States的動(dòng)態(tài)擴(kuò)展是非常靈活的，現(xiàn)提供了3種擴(kuò)展，下面分別介紹：

ListState:并發(fā)度在改變的時(shí)候，會(huì)將并發(fā)上的每個(gè)List都取出，然后把這些List合并到一個(gè)新的List,然后根據(jù)元素的個(gè)數(shù)在均勻分配給新的Task;

UnionListState:相比于ListState更加靈活，把劃分的方式交給用戶去做，當(dāng)改變并發(fā)的時(shí)候，會(huì)將原來(lái)的List拼接起來(lái)。然后不做劃分，直接交給用戶；

BroadcastState:如大表和小表做Join時(shí)，小表可以直接廣播給大表的分區(qū)，在每個(gè)并發(fā)上的數(shù)據(jù)都是完全一致的。做的更新也相同，當(dāng)改變并發(fā)的時(shí)候，把這些數(shù)據(jù)COPY到新的Task即可

以上是Flink Operator States提供的3種擴(kuò)展方式，用戶可以根據(jù)自己的需求做選擇。

使用Checkpoint提高程序的可靠性

用戶可以根據(jù)的程序里面的配置將checkpoint打開(kāi)，給定一個(gè)時(shí)間間隔后，框架會(huì)按照時(shí)間間隔給程序的狀態(tài)進(jìn)行備份。當(dāng)發(fā)生故障時(shí)，F(xiàn)link會(huì)將所有Task的狀態(tài)一起恢復(fù)到Checkpoint的狀態(tài)。從哪個(gè)位置開(kāi)始重新執(zhí)行。

Flink也提供了多種正確性的保障，包括：

AT LEAST ONCE;

Exactly once;

備份為保存在State中的程序狀態(tài)數(shù)據(jù)

Flink也提供了一套機(jī)制，允許把這些狀態(tài)放到內(nèi)存當(dāng)中。做Checkpoint的時(shí)候，由Flink去完成恢復(fù)。

從已停止作業(yè)的運(yùn)行狀態(tài)中恢復(fù)

當(dāng)組件升級(jí)的時(shí)候，需要停止當(dāng)前作業(yè)。這個(gè)時(shí)候需要從之前停止的作業(yè)當(dāng)中恢復(fù)，F(xiàn)link提供了2種機(jī)制恢復(fù)作業(yè):

Savepoint:是一種特殊的checkpoint，只不過(guò)不像checkpoint定期的從系統(tǒng)中去觸發(fā)的，它是用戶通過(guò)命令觸發(fā)，存儲(chǔ)格式和checkpoint也是不相同的，會(huì)將數(shù)據(jù)按照一個(gè)標(biāo)準(zhǔn)的格式存儲(chǔ)，不管配置什么樣，F(xiàn)link都會(huì)從這個(gè)checkpoint恢復(fù)，是用來(lái)做版本升級(jí)一個(gè)非常好的工具；

External Checkpoint：對(duì)已有checkpoint的一種擴(kuò)展，就是說(shuō)做完一次內(nèi)部的一次Checkpoint后，還會(huì)在用戶給定的一個(gè)目錄中，多存儲(chǔ)一份checkpoint的數(shù)據(jù)；

下面介紹一下?tīng)顟B(tài)管理和容錯(cuò)機(jī)制實(shí)現(xiàn)方式，F(xiàn)link提供了3種不同的StateBackend

MemoryStateBackend

FsStateBackend

RockDBStateBackend

用戶可以根據(jù)自己的需求選擇，如果數(shù)據(jù)量較小，可以存放到MemoryStateBackend和FsStateBackend中，如果數(shù)據(jù)量較大，可以放到RockDB中。

下面介紹HeapKeyedStateBackend和RockDBKeyedStateBackend

第一，HeapKeyedStateBackend

第二，RockDBKeyedStateBackend

Checkpoint的執(zhí)行流程

Checkpoint的執(zhí)行流程是按照Chandy-Lamport算法實(shí)現(xiàn)的。

Checkpoint Barrier的對(duì)齊

全量Checkpoint

全量Checkpoint會(huì)在每個(gè)節(jié)點(diǎn)做備份數(shù)據(jù)時(shí)，只需要將數(shù)據(jù)都便利一遍，然后寫(xiě)到外部存儲(chǔ)中，這種情況會(huì)影響備份性能。在此基礎(chǔ)上做了優(yōu)化。

RockDB的數(shù)據(jù)會(huì)更新到內(nèi)存，當(dāng)內(nèi)存滿時(shí)，會(huì)寫(xiě)入到磁盤(pán)中。增量的機(jī)制會(huì)將新產(chǎn)生的文件COPY持久化中，而之前產(chǎn)生的文件就不需要COPY到持久化中去了。通過(guò)這種方式減少COPY的數(shù)據(jù)量，并提高性能。

阿里是從2015年開(kāi)始調(diào)研Flink,2015年10月啟動(dòng)Blink項(xiàng)目，并完善Flink在大規(guī)模生產(chǎn)下的一些優(yōu)化和改進(jìn)。2016年雙11采用了Blink系統(tǒng)，為搜索，推薦，廣告業(yè)務(wù)提供服務(wù)。2017年5月Blink已成為阿里的實(shí)時(shí)計(jì)算引擎。

正在做的工作，基于State重構(gòu)Window方面的一些優(yōu)化，阿里也正在將功能做完善。后續(xù)將包括asynchronous Checkpoint的功能完善，并和社區(qū)進(jìn)一步溝通和合作。幫助Flink社區(qū)完善相關(guān)方面的工作。

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