目錄

• System Architecture
• Data Transfer in Flink
• Event Time Processing
• State Management
• Checkpoints, Savepoints, and State Recovery

System Architecture

分布式系統(tǒng)需要解決：分配和管理在集群的計(jì)算資源、處理配合码秉、持久和可訪問(wèn)的數(shù)據(jù)存儲(chǔ)府蔗、失敗恢復(fù)。Fink專注分布式流處理蝌焚。

Components of a Flink Setup

• JobManager ：接受application岖食，包含StreamGraph（DAG）蔑穴、JobGraph（logical dataflow graph，已經(jīng)進(jìn)過(guò)優(yōu)化纵朋，如task chain）和JAR聂薪，將JobGraph轉(zhuǎn)化為ExecutionGraph（physical dataflow graph，并行化）复罐，包含可以并發(fā)執(zhí)行的tasks乱投。其他工作類似Spark driver钮惠，如向RM申請(qǐng)資源素挽、schedule tasks究西、保存作業(yè)的元數(shù)據(jù)，如checkpoints较屿。如今JM可分為JobMaster和ResourceManager（和下面的不同），分別負(fù)責(zé)任務(wù)和資源，在Session模式下啟動(dòng)多個(gè)job就會(huì)有多個(gè)JobMaster。
• ResourceManager：一般是Yarn，當(dāng)TM有空閑的slot就會(huì)告訴JM妆距，沒(méi)有足夠的slot也會(huì)啟動(dòng)新的TM。kill掉長(zhǎng)時(shí)間空閑的TM屈芜。
• TaskManager類似Spark的executor躬翁，會(huì)跑多個(gè)線程的task、數(shù)據(jù)緩存與交換蛮艰。
• Dispatcher（Application Master）提供REST接口來(lái)接收client的application提交仍律，它負(fù)責(zé)啟動(dòng)JM和提交application钢拧，同時(shí)運(yùn)行Web UI膜钓。

task是最基本的調(diào)度單位，由一個(gè)線程執(zhí)行，里面包含一個(gè)或多個(gè)operator。多個(gè)operators就成為operation chain炒辉，需要上下游并發(fā)度一致屏轰，且傳遞模式（之前的Data exchange strategies）是forward唁盏。

slot是TM的資源子集昂灵。結(jié)合下面Task Execution的圖佩谣，一個(gè)slot并不代表一個(gè)線程艇炎，它里面并不一定只放一個(gè)task。多個(gè)task在一個(gè)slot就涉及slot sharing group疆柔。一個(gè)jobGraph的任務(wù)需要多少slot，取決于最大的并發(fā)度，這樣的話熙含，并發(fā)1和并發(fā)2就不會(huì)放到一個(gè)slot中。Co-Location Group是在此基礎(chǔ)上塘装，數(shù)據(jù)的forward形式裂七，即一個(gè)slot中毛雇，如果它處理的是key1的數(shù)據(jù)，那么接下來(lái)的task也是處理key1的數(shù)據(jù)羡棵，此時(shí)就達(dá)到Co-Location Group壁查。

盡管有slot sharing group，但一個(gè)group里串聯(lián)起來(lái)的task各自所需資源的大小并不好確定。阿里日常用得最多的還是一個(gè)task一個(gè)slot的方式。

image

Session模式（上圖）：預(yù)先啟動(dòng)好AM和TM，每提交一個(gè)job就啟動(dòng)一個(gè)Job Manager并向Flink的RM申請(qǐng)資源悠汽，不夠的話慨亲，F(xiàn)link的RM向YARN的RM申請(qǐng)資源刑棵。適合規(guī)模小胡陪，運(yùn)行時(shí)間短的作業(yè)。./bin/flink run ./path/to/job.jar

Job模式：每一個(gè)job都重新啟動(dòng)一個(gè)Flink集群吹泡，完成后結(jié)束Flink，且只有一個(gè)Job Manager趟径。資源按需申請(qǐng)幕屹，適合大作業(yè)蓝丙。./bin/flink run -m yarn-cluster ./path/to/job.jar

下面是簡(jiǎn)單例子，詳細(xì)看官網(wǎng)望拖。

# 啟動(dòng)yarn-session渺尘，4個(gè)TM，每個(gè)有4GB堆內(nèi)存说敏，4個(gè)slot
cd flink-1.7.0/
./bin/yarn-session.sh -n 4 -jm 1024m -tm 4096m -s 4
# 啟動(dòng)作業(yè)
./bin/flink run -m yarn-cluster -yn 4 -yjm 1024m -ytm 4096m ./examples/batch/WordCount.jar

細(xì)節(jié)取決于具體環(huán)境鸥跟，如不同的RM

Application Deployment

Framework模式：Flink作業(yè)為JAR，并被提交到Dispatcher or JM or YARN盔沫。

Library模式：Flink作業(yè)為application-specific container image医咨，如Docker image，適合微服務(wù)架诞。

Task Execution

作業(yè)調(diào)度：在流計(jì)算中預(yù)先啟動(dòng)好節(jié)點(diǎn)拟淮，而在批計(jì)算中，每當(dāng)某個(gè)階段完成計(jì)算才啟動(dòng)下一個(gè)節(jié)點(diǎn)谴忧。

資源管理：slot作為基本單位惩歉，有大小和位置屬性。JM有SlotPool俏蛮，向Flink RM申請(qǐng)Slot，F(xiàn)linkRM發(fā)現(xiàn)自己的SlotManager中沒(méi)有足夠的Slot上遥，就會(huì)向集群RM申請(qǐng)搏屑。后者返回可用TM的ip，讓FlinkRM去啟動(dòng)粉楚，TM啟動(dòng)后向FlinkRM注冊(cè)辣恋。后者向TM請(qǐng)求Slot，TM向JM提供相應(yīng)Slot模软。JM用完后釋放Slot伟骨，TM會(huì)把釋放的Slot報(bào)告給FlinkRM。在Blink版本中燃异，job模式會(huì)根據(jù)申請(qǐng)slot的大小分配相應(yīng)的TM携狭，而session模式則預(yù)先設(shè)置好TM大小，每有slot申請(qǐng)就從TM中劃分相應(yīng)的資源回俐。

任務(wù)可以是相同operator (data parallelism)逛腿，不同 operator (task parallelism)稀并，甚至不同application (job parallelism)。TM提供一定數(shù)量的slots來(lái)控制并行的任務(wù)數(shù)单默。

image

上圖A和C是source function碘举，E是sink function，小數(shù)字表示并行度搁廓。

image

一個(gè)TM是一個(gè)JVM進(jìn)程引颈，它通過(guò)多線程完成任務(wù)。線程的隔離不太好境蜕，一個(gè)線程失敗有可能導(dǎo)致整個(gè)TM失敗蝙场。

Highly-Available Setup

從失敗中恢復(fù)需要重啟失敗進(jìn)程、作業(yè)和恢復(fù)它的state汽摹。

當(dāng)一個(gè)TM掛掉而RM又無(wú)法找到空閑的資源時(shí)李丰，就只能暫時(shí)降低并行度，直到有空閑的資源重啟TM逼泣。

當(dāng)JM掛掉就靠ZK來(lái)重新選舉趴泌，和找到JM存儲(chǔ)到遠(yuǎn)程storage的元數(shù)據(jù)、JobGraph拉庶。重啟JM并從最后一個(gè)完成的checkpoint開始嗜憔。

JM在執(zhí)行期間會(huì)得到每個(gè)task checkpoints的state存儲(chǔ)路徑（task將state寫到遠(yuǎn)程storage）并寫到遠(yuǎn)程storage，同時(shí)在ZK的存儲(chǔ)路徑留下pointer指明到哪里找上面的存儲(chǔ)路徑氏仗。

image

背壓

數(shù)據(jù)涌入的速度大于處理速度吉捶。在source operator中，可通過(guò)Kafka解決皆尔。在任務(wù)間的operator有如下機(jī)制應(yīng)對(duì)：

Local exchange：task1和2在同一個(gè)工作節(jié)點(diǎn)呐舔，那么buffer pool可以直接交給下一個(gè)任務(wù)，但下一個(gè)任務(wù)task2消費(fèi)buffer pool中的信息速度減慢時(shí)慷蠕，當(dāng)前任務(wù)task1填充buffer pool的速度也會(huì)減慢珊拼。

Remote exchange：TM保證每個(gè)task至少有一個(gè)incoming和一個(gè)outgoing緩沖區(qū)。當(dāng)下游receiver的處理速度低于上有的sender的發(fā)送速度流炕，receiver的incoming緩沖區(qū)就會(huì)開始積累數(shù)據(jù)（需要空閑的buffer來(lái)放從TCP連接中接收的數(shù)據(jù)）澎现，當(dāng)擠滿后就不再接收數(shù)據(jù)。上游sender利用netty水位機(jī)制每辟，當(dāng)網(wǎng)絡(luò)中的緩沖數(shù)據(jù)過(guò)多時(shí)暫停發(fā)送剑辫。

Data Transfer in Flink

TM負(fù)責(zé)數(shù)據(jù)在tasks間的轉(zhuǎn)移，轉(zhuǎn)移之前會(huì)存儲(chǔ)到buffer（這又變回micro-batches）渠欺。每個(gè)TM有32KB的網(wǎng)絡(luò)buffer用于接收和發(fā)送數(shù)據(jù)妹蔽。如果sender和receiver在不同進(jìn)程，那么會(huì)通過(guò)操作系統(tǒng)的網(wǎng)絡(luò)棧來(lái)通信。每對(duì)TM保持permanent TCP連接來(lái)交換數(shù)據(jù)讹开。每個(gè)sender任務(wù)能夠給所有receiving任務(wù)發(fā)送數(shù)據(jù)盅视，反之，所有receiver任務(wù)能夠接收所有sender任務(wù)的數(shù)據(jù)旦万。TM保證每個(gè)任務(wù)都至少有一個(gè)incoming和outgoing的buffer闹击，并增加額外的緩沖區(qū)分配約束來(lái)避免死鎖。

image

如果sender和receiver任務(wù)在同一個(gè)TM進(jìn)程成艘，sender會(huì)序列化結(jié)果數(shù)據(jù)到buffer赏半，如果滿了就放到隊(duì)列。receiver任務(wù)通過(guò)隊(duì)列得到數(shù)據(jù)并進(jìn)行反序列化淆两。這樣的好處是解耦任務(wù)并允許在任務(wù)中使用可變對(duì)象断箫，從而減少了對(duì)象實(shí)例化和垃圾收集。一旦數(shù)據(jù)被序列化秋冰，就能安全地修改仲义。而缺點(diǎn)是計(jì)算消耗大，在一些條件下能夠把task穿起來(lái)剑勾，避免序列化埃撵。(C10)

Flow Control with Back Pressure

receiver放到緩沖區(qū)的數(shù)據(jù)變?yōu)殛?duì)列，sender將要發(fā)送的數(shù)據(jù)變?yōu)殛?duì)列虽另，最后sender減慢發(fā)送速度暂刘。

Event Time Processing

event time處理的數(shù)據(jù)必須有時(shí)間戳（Long unix timestamp）并定義了watermarks。watermark是一種特殊的records holding a timestamp long value捂刺。它必須是遞增的（防止倒退）谣拣，有一個(gè)timestamp t（下圖的5），暗示所有接下來(lái)的數(shù)據(jù)都會(huì)大于這個(gè)值族展。后來(lái)的森缠，小于這個(gè)值，就被視為遲來(lái)數(shù)據(jù)仪缸，F(xiàn)link有其他機(jī)制處理辅鲸。

image

Watermarks and Event Time

WM在Flink是一種特殊的record，它會(huì)被operator tasks接收和釋放腹殿。

tasks有時(shí)間服務(wù)來(lái)維持timers（timers注冊(cè)到時(shí)間服務(wù)上），在time-window task中例书，timers分別記錄了各個(gè)window的結(jié)束時(shí)間锣尉。當(dāng)任務(wù)獲得一個(gè)watermark時(shí)，task會(huì)根據(jù)這個(gè)watermark的timestamp更新內(nèi)部的event-time clock决采。任務(wù)內(nèi)部的時(shí)間服務(wù)確定所有timers時(shí)間是否小于watermark的timestamp自沧，如果大于則觸發(fā)call-back算子來(lái)釋放記錄并返回結(jié)果。最后task還會(huì)將更新的event-time clock的WM進(jìn)行廣播。（結(jié)合下圖理解）

只有ProcessFunction可以讀取和修改timestamp或者watermark（The ProcessFunction can read the timestamp of a currently processed record, request the current event-time of the operator, and register timers）拇厢。下面是PF的行為爱谁。

image

當(dāng)收到WM大于所有目前擁有的WM，就會(huì)把event-time clock更新為所有WM中最小的那個(gè)孝偎，并廣播這個(gè)最小的WM访敌。即便是多個(gè)streams輸入，機(jī)制也一樣衣盾，只是增加Paritition WM數(shù)量寺旺。這種機(jī)制要求獲得的WM必須是累加的，而且task必須有新的WM接收势决，否則clock就不會(huì)更新阻塑，task的timers就不會(huì)被觸發(fā)。另外果复，當(dāng)多個(gè)streams輸入時(shí)陈莽，timers會(huì)被WM比較離散的stream主導(dǎo)，從而使更密集的stream的state不斷積累虽抄。

Timestamp Assignment and Watermark Generation

當(dāng)streaming application消化流時(shí)產(chǎn)生走搁。Flink有三種方式產(chǎn)生：

• SourceFunction：產(chǎn)生的record帶有timestamp，一些特殊時(shí)點(diǎn)產(chǎn)生WM极颓。如果SF暫時(shí)不再發(fā)送WM朱盐，則會(huì)被認(rèn)為是idle。Flink會(huì)從接下來(lái)的watermark operators中排除由這個(gè)SF生產(chǎn)的分區(qū)（上圖有4個(gè)分區(qū)）菠隆，從而解決timer不觸發(fā)的問(wèn)題兵琳。
• AssignerWithPeriodicWatermarks 提取每條記錄的timestamp，并周期性的查詢當(dāng)前WM骇径，即上圖的Partition WM躯肌。
• AssignerWithPunctuatedWatermarks 可以從每條數(shù)據(jù)提取WM。

上面兩個(gè)User-defined timestamp assignment functions通常用在source operator附近破衔，因?yàn)閟tream一經(jīng)處理就很難把握record的時(shí)間順序了清女。所以UDF可以修改timestamp和WM，但在數(shù)據(jù)處理時(shí)使用不是一個(gè)好主意晰筛。

State Management

由任務(wù)維護(hù)并用于計(jì)算函數(shù)結(jié)果的所有數(shù)據(jù)都屬于任務(wù)的state嫡丙。其實(shí)state可以理解為task業(yè)務(wù)邏輯的本地或?qū)嵗兞俊?/p>

image

在Flink，state總是和特定的operator關(guān)聯(lián)读第。operator需要注冊(cè)它的state曙博，而state有兩種類型：

• Operator State：由同一并行任務(wù)處理的所有記錄都可以訪問(wèn)相同的state，而其他的task或operator不能訪問(wèn)怜瞒，即一個(gè)task專屬一個(gè)state父泳。這種state有三種primitives
  • List State represents state as a list of entries.
  • Union List State同上，但在任務(wù)失敗和作業(yè)從savepoint重啟的行為不一樣
  • Broadcast State（v1.5） 同樣一個(gè)task專屬一個(gè)state，但state都是一樣的（需要自己注意保持一致惠窄，對(duì)state更新時(shí)蒸眠，實(shí)際上只對(duì)當(dāng)前task的state進(jìn)行更新。只有所有task的更新一樣時(shí)杆融，即輸入數(shù)據(jù)一樣（一開始廣播所以一樣楞卡，但數(shù)據(jù)的順序可能不一樣），對(duì)數(shù)據(jù)的處理一樣擒贸，才能保證state一樣）臀晃。這種state只能存儲(chǔ)在內(nèi)存，所以沒(méi)有RockDB backend介劫。
• Keyed State：相同key的record共享一個(gè)state徽惋。
  • Value State：每個(gè)key一個(gè)值，這個(gè)值可以是復(fù)雜的數(shù)據(jù)結(jié)構(gòu).
  • List State：每個(gè)key一個(gè)list
  • Map State：每個(gè)key一個(gè)map

上面兩種state的存在方式有兩種：raw和managed座韵，一般都是用后者险绘，也推薦用后者（更好的內(nèi)存管理、不需造輪子）誉碴。

State Backends

state backend決定了state如何被存儲(chǔ)宦棺、訪問(wèn)和維持。它的主要職責(zé)是本地state管理和checkpoint state到遠(yuǎn)程黔帕。在管理方面代咸，可選擇將state存儲(chǔ)到內(nèi)存還是磁盤。checkpoint方面在C8詳細(xì)介紹成黄。

MemoryStateBackend, FsStateBackend, RocksDBStateBackend適合越來(lái)越大的state呐芥。都支持異步checkpoint，其中RocksDB還支持incremental的checkpoint奋岁。

• 注意：As RocksDB’s JNI bridge API is based on byte[], the maximum supported size per key and per value is 2^31 bytes each. IMPORTANT: states that use merge operations in RocksDB (e.g. ListState) can silently accumulate value sizes > 2^31 bytes and will then fail on their next retrieval. This is currently a limitation of RocksDB JNI.

Scaling Stateful Operators

Flink會(huì)根據(jù)input rate調(diào)整并發(fā)度思瘟。對(duì)于stateful operators有以下4種方式：

• keyed state：根據(jù)key group來(lái)調(diào)整，即分為同一組的key-value會(huì)被分到相同的task

• list state：所有l(wèi)ist entries會(huì)被收集并重新均勻分布闻伶，當(dāng)增加并發(fā)度時(shí)滨攻，要新建list

• union list state：增加并發(fā)時(shí)，廣播整個(gè)list蓝翰，所以rescaling后光绕，所有task都有所有的list state。

image

• broadcast state

image

Checkpoints, Savepoints, and State Recovery

Flink’s Lightweight Checkpointing Algorithm

在分布式開照算法Chandy-Lamport的基礎(chǔ)上實(shí)現(xiàn)畜份。有一種特殊的record叫checkpoint barrier（由JM產(chǎn)生）奇钞，它帶有checkpoint ID來(lái)把流進(jìn)行劃分。在CB前面的records會(huì)被包含到checkpoint漂坏，之后的會(huì)被包含在之后的checkpoint。

當(dāng)source task收到這種信息，就會(huì)停止發(fā)送recordes顶别，觸發(fā)state backend對(duì)本地state的checkpoint谷徙，并廣播checkpoint ID到所有下游task。當(dāng)checkpoint完成時(shí)驯绎，state backend喚醒source task完慧，后者向JM確定相應(yīng)的checkpoint ID已經(jīng)完成任務(wù)。

image

當(dāng)下游獲得其中一個(gè)CB時(shí)剩失，就會(huì)暫停處理這個(gè)CB對(duì)應(yīng)的source的數(shù)據(jù)（完成checkpoint后發(fā)送的數(shù)據(jù)）屈尼，并將這些數(shù)據(jù)存到緩沖區(qū)，直到其他相同ID的CB都到齊拴孤，就會(huì)把state（下圖的12脾歧、8）進(jìn)行checkpoint，并廣播CB到下游演熟。直到所有CB被廣播到下游衰抑，才開始處理排隊(duì)在緩沖區(qū)的數(shù)據(jù)劫哼。當(dāng)然，其他沒(méi)有發(fā)送CB的source的數(shù)據(jù)會(huì)繼續(xù)處理。

image

image

最后劫乱，當(dāng)所有sink會(huì)向JM發(fā)送BC確定checkpoint已完成。

這種機(jī)制還有兩個(gè)優(yōu)化：

• 當(dāng)operator的state很大時(shí)拗军，復(fù)制整個(gè)state并發(fā)送到遠(yuǎn)程storage會(huì)很費(fèi)時(shí)殊轴。而RocksDB state backend支持asynchronous and incremental的checkpoints。當(dāng)觸發(fā)checkpoint時(shí)座云，backend會(huì)快照所有本地state的修改（直至上一次checkpoint）疙赠，然后馬上讓task繼續(xù)執(zhí)行。后臺(tái)線程異步發(fā)送快照到遠(yuǎn)程storage疙教。
• 在等待其余CB時(shí)棺聊，已經(jīng)完成checkpoint的source數(shù)據(jù)需要排隊(duì)。但如果使用at-least-once就不需要等了贞谓。但當(dāng)所有CB到齊再checkpoint限佩，存儲(chǔ)的state就已經(jīng)包含了下一次checkpoint才記錄的數(shù)據(jù)。（如果是取最值這種state就無(wú)所謂）

Recovery from Consistent Checkpoints

image

上圖隊(duì)列中的7和6之所以能恢復(fù)裸弦，取決于數(shù)據(jù)源是否resettable祟同，如Kafka，不會(huì)因?yàn)榘l(fā)送信息就把信息刪除理疙。這才能實(shí)現(xiàn)處理過(guò)程的exactly-once state consistency（嚴(yán)格來(lái)講晕城，數(shù)據(jù)還是被重復(fù)處理，但是在讀檔后重復(fù)的）窖贤。但是下游系統(tǒng)有可能接收到多個(gè)結(jié)果砖顷。這方面贰锁，F(xiàn)link提供sink算子實(shí)現(xiàn)output的exactly-once，例如給checkpoint提交records釋放記錄滤蝠。另一個(gè)方法是idempotent updates豌熄，詳細(xì)看C7。

Savepoints

checkpoints加上一些額外的元數(shù)據(jù)物咳，功能也是在checkpoint的基礎(chǔ)上豐富锣险。不同于checkpoints，savepoint不會(huì)被Flink自動(dòng)創(chuàng)造（由用戶或者外部scheduler觸發(fā)創(chuàng)造）和銷毀览闰。savepoint可以重啟不同但兼容的作業(yè)芯肤，從而：

• 修復(fù)bugs進(jìn)而修復(fù)錯(cuò)誤的結(jié)果，也可用于A/B test或者what-if場(chǎng)景压鉴。
• 調(diào)整并發(fā)度
• 遷移作業(yè)到其他集群崖咨、新版Flink

也可以用于暫停作業(yè)，通過(guò)savepoint查看作業(yè)情況晴弃。

參考
Stream Processing with Apache Flink by Vasiliki Kalavri; Fabian Hueske

原文：https://www.cnblogs.com/code2one/p/10123112.html

Kotlin 開發(fā)者社區(qū)

國(guó)內(nèi)第一Kotlin 開發(fā)者社區(qū)公眾號(hào)掩幢，主要分享、交流 Kotlin 編程語(yǔ)言上鞠、Spring Boot际邻、Android、React.js/Node.js芍阎、函數(shù)式編程世曾、編程思想等相關(guān)主題。

越是喧囂的世界谴咸，越需要寧?kù)o的思考轮听。