System Architecture

分布式系統(tǒng)需要解決：分配和管理在集群的計算資源、處理配合突想、持久和可訪問的數(shù)據(jù)存儲殴蹄、失敗恢復(fù)。Fink專注分布式流處理猾担。

Components of a Flink Setup

JobManager ：接受application袭灯，包含StreamGraph（DAG）、JobGraph（logical dataflow graph垒探，已經(jīng)進(jìn)過優(yōu)化妓蛮，如task chain）和JAR，將JobGraph轉(zhuǎn)化為ExecutionGraph（physical dataflow graph圾叼，并行化）蛤克，包含可以并發(fā)執(zhí)行的tasks。其他工作類似Spark driver夷蚊，如向RM申請資源构挤、schedule tasks、保存作業(yè)的元數(shù)據(jù)惕鼓，如checkpoints筋现。如今JM可分為JobMaster和ResourceManager（和下面的不同），分別負(fù)責(zé)任務(wù)和資源箱歧，在Session模式下啟動多個job就會有多個JobMaster矾飞。

ResourceManager：一般是Yarn，當(dāng)TM有空閑的slot就會告訴JM呀邢，沒有足夠的slot也會啟動新的TM洒沦。kill掉長時間空閑的TM。

TaskManager類似Spark的executor价淌，會跑多個線程的task申眼、數(shù)據(jù)緩存與交換瞒津。

Dispatcher（Application Master）提供REST接口來接收client的application提交，它負(fù)責(zé)啟動JM和提交application括尸，同時運行Web UI巷蚪。

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

slot是TM的資源子集功戚。結(jié)合下面Task Execution的圖娶眷，一個slot并不代表一個線程，它里面并不一定只放一個task啸臀。多個task在一個slot就涉及slot sharing group届宠。一個jobGraph的任務(wù)需要多少slot，取決于最大的并發(fā)度乘粒，這樣的話豌注，并發(fā)1和并發(fā)2就不會放到一個slot中。Co-Location Group是在此基礎(chǔ)上灯萍，數(shù)據(jù)的forward形式轧铁，即一個slot中，如果它處理的是key1的數(shù)據(jù)旦棉，那么接下來的task也是處理key1的數(shù)據(jù)齿风，此時就達(dá)到Co-Location Group。

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

Session模式（上圖）：預(yù)先啟動好AM和TM真屯，每提交一個job就啟動一個Job Manager并向Flink的RM申請資源脸候，不夠的話，F(xiàn)link的RM向YARN的RM申請資源绑蔫。適合規(guī)模小运沦，運行時間短的作業(yè)。./bin/flink run ./path/to/job.jar

Job模式：每一個job都重新啟動一個Flink集群配深，完成后結(jié)束Flink携添，且只有一個Job Manager。資源按需申請凉馆，適合大作業(yè)薪寓。./bin/flink run -m yarn-cluster ./path/to/job.jar

下面是簡單例子亡资，詳細(xì)看官網(wǎng)。

# 啟動yarn-session向叉，4個TM锥腻，每個有4GB堆內(nèi)存，4個slotcdflink-1.7.0/./bin/yarn-session.sh -n 4 -jm 1024m -tm 4096m -s 4# 啟動作業(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)度：在流計算中預(yù)先啟動好節(jié)點咬扇，而在批計算中甲葬，每當(dāng)某個階段完成計算才啟動下一個節(jié)點。

資源管理：slot作為基本單位懈贺，有大小和位置屬性经窖。JM有SlotPool，向Flink RM申請Slot梭灿，F(xiàn)linkRM發(fā)現(xiàn)自己的SlotManager中沒有足夠的Slot画侣，就會向集群RM申請。后者返回可用TM的ip堡妒，讓FlinkRM去啟動配乱，TM啟動后向FlinkRM注冊。后者向TM請求Slot皮迟，TM向JM提供相應(yīng)Slot搬泥。JM用完后釋放Slot，TM會把釋放的Slot報告給FlinkRM万栅。在Blink版本中佑钾，job模式會根據(jù)申請slot的大小分配相應(yīng)的TM，而session模式則預(yù)先設(shè)置好TM大小烦粒，每有slot申請就從TM中劃分相應(yīng)的資源休溶。

任務(wù)可以是相同operator (data parallelism)，不同 operator (task parallelism)扰她，甚至不同application (job parallelism)兽掰。TM提供一定數(shù)量的slots來控制并行的任務(wù)數(shù)。

上圖A和C是source function徒役，E是sink function孽尽，小數(shù)字表示并行度。

一個TM是一個JVM進(jìn)程忧勿，它通過多線程完成任務(wù)杉女。線程的隔離不太好瞻讽，一個線程失敗有可能導(dǎo)致整個TM失敗。

Highly-Available Setup

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

當(dāng)一個TM掛掉而RM又無法找到空閑的資源時，就只能暫時降低并行度坎拐，直到有空閑的資源重啟TM烦磁。

當(dāng)JM掛掉就靠ZK來重新選舉，和找到JM存儲到遠(yuǎn)程storage的元數(shù)據(jù)哼勇、JobGraph都伪。重啟JM并從最后一個完成的checkpoint開始。

JM在執(zhí)行期間會得到每個task checkpoints的state存儲路徑（task將state寫到遠(yuǎn)程storage）并寫到遠(yuǎn)程storage积担，同時在ZK的存儲路徑留下pointer指明到哪里找上面的存儲路徑陨晶。

背壓

數(shù)據(jù)涌入的速度大于處理速度。在source operator中磅轻，可通過Kafka解決珍逸。在任務(wù)間的operator有如下機制應(yīng)對：

Local exchange：task1和2在同一個工作節(jié)點逐虚，那么buffer pool可以直接交給下一個任務(wù)聋溜，但下一個任務(wù)task2消費buffer pool中的信息速度減慢時，當(dāng)前任務(wù)task1填充buffer pool的速度也會減慢叭爱。

Remote exchange：TM保證每個task至少有一個incoming和一個outgoing緩沖區(qū)撮躁。當(dāng)下游receiver的處理速度低于上有的sender的發(fā)送速度，receiver的incoming緩沖區(qū)就會開始積累數(shù)據(jù)（需要空閑的buffer來放從TCP連接中接收的數(shù)據(jù)）买雾，當(dāng)擠滿后就不再接收數(shù)據(jù)把曼。上游sender利用netty水位機制，當(dāng)網(wǎng)絡(luò)中的緩沖數(shù)據(jù)過多時暫停發(fā)送漓穿。

Data Transfer in Flink

TM負(fù)責(zé)數(shù)據(jù)在tasks間的轉(zhuǎn)移嗤军，轉(zhuǎn)移之前會存儲到buffer（這又變回micro-batches）。每個TM有32KB的網(wǎng)絡(luò)buffer用于接收和發(fā)送數(shù)據(jù)晃危。如果sender和receiver在不同進(jìn)程叙赚，那么會通過操作系統(tǒng)的網(wǎng)絡(luò)棧來通信。每對TM保持permanent TCP連接來交換數(shù)據(jù)僚饭。每個sender任務(wù)能夠給所有receiving任務(wù)發(fā)送數(shù)據(jù)震叮，反之，所有receiver任務(wù)能夠接收所有sender任務(wù)的數(shù)據(jù)鳍鸵。TM保證每個任務(wù)都至少有一個incoming和outgoing的buffer苇瓣，并增加額外的緩沖區(qū)分配約束來避免死鎖。

如果sender和receiver任務(wù)在同一個TM進(jìn)程偿乖，sender會序列化結(jié)果數(shù)據(jù)到buffer击罪，如果滿了就放到隊列哲嘲。receiver任務(wù)通過隊列得到數(shù)據(jù)并進(jìn)行反序列化。這樣的好處是解耦任務(wù)并允許在任務(wù)中使用可變對象媳禁，從而減少了對象實例化和垃圾收集撤蚊。一旦數(shù)據(jù)被序列化，就能安全地修改损话。而缺點是計算消耗大侦啸，在一些條件下能夠把task穿起來，避免序列化丧枪。(C10)

Flow Control with Back Pressure

receiver放到緩沖區(qū)的數(shù)據(jù)變?yōu)殛犃泄馔浚瑂ender將要發(fā)送的數(shù)據(jù)變?yōu)殛犃校詈髎ender減慢發(fā)送速度拧烦。

Event Time Processing

event time處理的數(shù)據(jù)必須有時間戳（Long unix timestamp）并定義了watermarks忘闻。watermark是一種特殊的records holding a timestamp long value。它必須是遞增的（防止倒退）恋博，有一個timestamp t（下圖的5）齐佳，暗示所有接下來的數(shù)據(jù)都會大于這個值。后來的债沮，小于這個值炼吴，就被視為遲來數(shù)據(jù)，F(xiàn)link有其他機制處理疫衩。

Watermarks and Event Time

WM在Flink是一種特殊的record硅蹦，它會被operator tasks接收和釋放。

tasks有時間服務(wù)來維持timers（timers注冊到時間服務(wù)上）闷煤，在time-window task中童芹，timers分別記錄了各個window的結(jié)束時間。當(dāng)任務(wù)獲得一個watermark時鲤拿，task會根據(jù)這個watermark的timestamp更新內(nèi)部的event-time clock假褪。任務(wù)內(nèi)部的時間服務(wù)確定所有timers時間是否小于watermark的timestamp，如果大于則觸發(fā)call-back算子來釋放記錄并返回結(jié)果近顷。最后task還會將更新的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的行為幕庐。

當(dāng)收到WM大于所有目前擁有的WM久锥，就會把event-time clock更新為所有WM中最小的那個，并廣播這個最小的WM异剥。即便是多個streams輸入瑟由，機制也一樣，只是增加Paritition WM數(shù)量。這種機制要求獲得的WM必須是累加的歹苦，而且task必須有新的WM接收青伤，否則clock就不會更新，task的timers就不會被觸發(fā)殴瘦。另外狠角，當(dāng)多個streams輸入時，timers會被WM比較離散的stream主導(dǎo)蚪腋，從而使更密集的stream的state不斷積累丰歌。

Timestamp Assignment and Watermark Generation

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

SourceFunction：產(chǎn)生的record帶有timestamp屉凯，一些特殊時點產(chǎn)生WM立帖。如果SF暫時不再發(fā)送WM，則會被認(rèn)為是idle悠砚。Flink會從接下來的watermark operators中排除由這個SF生產(chǎn)的分區(qū)（上圖有4個分區(qū)）晓勇，從而解決timer不觸發(fā)的問題。

AssignerWithPeriodicWatermarks?提取每條記錄的timestamp灌旧，并周期性的查詢當(dāng)前WM绑咱，即上圖的Partition WM。

AssignerWithPunctuatedWatermarks?可以從每條數(shù)據(jù)提取WM枢泰。

上面兩個User-defined timestamp assignment functions通常用在source operator附近描融，因為stream一經(jīng)處理就很難把握record的時間順序了。所以UDF可以修改timestamp和WM宗苍，但在數(shù)據(jù)處理時使用不是一個好主意稼稿。

State Management

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

在Flink讳窟，state總是和特定的operator關(guān)聯(lián)。operator需要注冊它的state敞恋，而state有兩種類型：

Operator State：由同一并行任務(wù)處理的所有記錄都可以訪問相同的state丽啡，而其他的task或operator不能訪問，即一個task專屬一個state硬猫。這種state有三種primitives

List State?represents state as a list of entries.

Union List State同上补箍，但在任務(wù)失敗和作業(yè)從savepoint重啟的行為不一樣

Broadcast State（v1.5） 同樣一個task專屬一個state尘喝，但state都是一樣的（需要自己注意保持一致校辩，對state更新時，實際上只對當(dāng)前task的state進(jìn)行更新惶洲。只有所有task的更新一樣時衬横，即輸入數(shù)據(jù)一樣（一開始廣播所以一樣裹粤，但數(shù)據(jù)的順序可能不一樣），對數(shù)據(jù)的處理一樣蜂林，才能保證state一樣）遥诉。這種state只能存儲在內(nèi)存拇泣，所以沒有RockDB backend。

Keyed State：相同key的record共享一個state矮锈。

Value State：每個key一個值霉翔，這個值可以是復(fù)雜的數(shù)據(jù)結(jié)構(gòu).

List State：每個key一個list

Map State：每個key一個map

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

State Backends

state backend決定了state如何被存儲瀑凝、訪問和維持葱弟。它的主要職責(zé)是本地state管理和checkpoint state到遠(yuǎn)程。在管理方面猜丹，可選擇將state存儲到內(nèi)存還是磁盤芝加。checkpoint方面在C8詳細(xì)介紹。

MemoryStateBackend, FsStateBackend, RocksDBStateBackend適合越來越大的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會根據(jù)input rate調(diào)整并發(fā)度脉顿。對于stateful operators有以下4種方式：

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

list state：所有l(wèi)ist entries會被收集并重新均勻分布，當(dāng)增加并發(fā)度時艾疟，要新建list

union list state：增加并發(fā)時来吩，廣播整個list，所以rescaling后蔽莱，所有task都有所有的list state弟疆。

broadcast state

Checkpoints, Savepoints, and State Recovery

Flink’s Lightweight Checkpointing Algorithm

在分布式開照算法Chandy-Lamport的基礎(chǔ)上實現(xiàn)。有一種特殊的record叫checkpoint barrier（由JM產(chǎn)生）盗冷，它帶有checkpoint ID來把流進(jìn)行劃分怠苔。在CB前面的records會被包含到checkpoint，之后的會被包含在之后的checkpoint仪糖。

當(dāng)source task收到這種信息柑司，就會停止發(fā)送recordes，觸發(fā)state backend對本地state的checkpoint锅劝，并廣播checkpoint ID到所有下游task攒驰。當(dāng)checkpoint完成時，state backend喚醒source task故爵，后者向JM確定相應(yīng)的checkpoint ID已經(jīng)完成任務(wù)玻粪。

當(dāng)下游獲得其中一個CB時，就會暫停處理這個CB對應(yīng)的source的數(shù)據(jù)（完成checkpoint后發(fā)送的數(shù)據(jù)），并將這些數(shù)據(jù)存到緩沖區(qū)奶段，直到其他相同ID的CB都到齊饥瓷，就會把state（下圖的12、8）進(jìn)行checkpoint痹籍，并廣播CB到下游呢铆。直到所有CB被廣播到下游，才開始處理排隊在緩沖區(qū)的數(shù)據(jù)蹲缠。當(dāng)然棺克，其他沒有發(fā)送CB的source的數(shù)據(jù)會繼續(xù)處理。

最后线定，當(dāng)所有sink會向JM發(fā)送BC確定checkpoint已完成娜谊。

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

當(dāng)operator的state很大時，復(fù)制整個state并發(fā)送到遠(yuǎn)程storage會很費時斤讥。而RocksDB state backend支持asynchronous and incremental的checkpoints纱皆。當(dāng)觸發(fā)checkpoint時，backend會快照所有本地state的修改（直至上一次checkpoint）芭商，然后馬上讓task繼續(xù)執(zhí)行派草。后臺線程異步發(fā)送快照到遠(yuǎn)程storage。

在等待其余CB時铛楣，已經(jīng)完成checkpoint的source數(shù)據(jù)需要排隊近迁。但如果使用at-least-once就不需要等了。但當(dāng)所有CB到齊再checkpoint簸州，存儲的state就已經(jīng)包含了下一次checkpoint才記錄的數(shù)據(jù)鉴竭。（如果是取最值這種state就無所謂）

Recovery from Consistent Checkpoints

上圖隊列中的7和6之所以能恢復(fù)，取決于數(shù)據(jù)源是否resettable岸浑，如Kafka搏存，不會因為發(fā)送信息就把信息刪除。這才能實現(xiàn)處理過程的exactly-once state consistency（嚴(yán)格來講助琐，數(shù)據(jù)還是被重復(fù)處理祭埂，但是在讀檔后重復(fù)的）。但是下游系統(tǒng)有可能接收到多個結(jié)果兵钮。這方面，F(xiàn)link提供sink算子實現(xiàn)output的exactly-once舌界，例如給checkpoint提交records釋放記錄掘譬。另一個方法是idempotent updates，詳細(xì)看C7呻拌。

Savepoints

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

修復(fù)bugs進(jìn)而修復(fù)錯誤的結(jié)果，也可用于A/B test或者what-if場景袜炕。

調(diào)整并發(fā)度

遷移作業(yè)到其他集群本谜、新版Flink

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