記錄一下個人看了一些Flink文章后的理解與個人關(guān)注點(diǎn)，目錄如下，
0. Overview
1. 基本概念
2. 并行Dataflow
3. 基本模塊
- JobManager
- TaskManagers
- Client
4. 組件棧
- Deployment層
- Runtime層
- API層
- Libraries層
5. 內(nèi)部原理
- 容錯機(jī)制
- 調(diào)度機(jī)制
- 迭代機(jī)制
- 反壓機(jī)制
6. Reference
Overview
基于Flink 1.4。
先來看看大數(shù)據(jù)計(jì)算引擎的發(fā)展路線，
- 第一代，hadoop的MapReduce
- 第二代，DAG框架的Tez，Oozie
- 第三代，Job內(nèi)部的DAG支持，以及強(qiáng)調(diào)實(shí)時計(jì)算，spark
- 第四代，迭代，流，批，SQL
基本概念
source -> transformation -> sink
- stream是算子的中間結(jié)果數(shù)據(jù)
- transformation是一個操作，它對一個或多個輸入stream進(jìn)行計(jì)算處理，輸出一個或多個結(jié)果stream
- streaming dataflow是一個執(zhí)行中的flink程序，啟動于一個或多個source，結(jié)束于一個或多個sink
并行Dataflow
一個stream可以被分成多個stream分區(qū)（stream partition）。
一個operator可以被分成多個operator subTask。
基本模塊
flink類似spark，是一個基于master-slave風(fēng)格的架構(gòu)。
運(yùn)行時runtime主要有2個進(jìn)程，一個是JobManagers，另一個是TaskManagers；client不屬于運(yùn)行時和程序執(zhí)行的一部分，而是用于準(zhǔn)備dataflow并將其發(fā)送到JobManager。
jobManager(master)是flink系統(tǒng)的協(xié)調(diào)者，負(fù)責(zé)接收flink job，調(diào)度組成job的多個task的執(zhí)行；手機(jī)job的狀態(tài)信息，管理flink集群中從節(jié)點(diǎn)taskManager，
- registerTaskManager，在Flink集群啟動的時候，TaskManager會向JobManager注冊
- submitJob，F(xiàn)link程序內(nèi)部通過Client向JobManager提交Flink Job，其中在消息SubmitJob中以JobGraph形式描述了Job的基本信息
- cancelJob，請求取消一個Flink Job的執(zhí)行，CancelJob消息中包含了Job的ID
- updateTaskExecutionState，TaskManager向JobManager請求更新狀態(tài)信息
- requestNextInputSplit，運(yùn)行在TaskManager上面的Task，請求獲取下一個要處理的輸入Split
- jobStatusChanged，表示Flink Job的狀態(tài)發(fā)生的變化
taskManager是一個actor(akka)，負(fù)責(zé)執(zhí)行計(jì)算的worker，在其上執(zhí)行flink job的一組task。每個taskManager負(fù)責(zé)管理其所在節(jié)點(diǎn)上的資源信息，如mem, disk, network，在啟動的時候?qū)①Y源狀態(tài)向jobManager匯報(bào)，
- 注冊階段，TaskManager會向JobManager注冊，發(fā)送registerTaskManager消息
- 可操作階段，接收并處理與Task有關(guān)的消息，如SubmitTask、CancelTask、FailTask
client，當(dāng)用戶提交一個flink程序時，會首先創(chuàng)建一個client，該client首先會對用戶提交的flink程序進(jìn)行預(yù)處理，并提交到flink集群中，
- client需要從用戶提交的flink程序配置中獲取jobManager的地址，并建立到j(luò)obManager的連接，將flink job提交給jobManager
- client會將用戶提交的flink程序組裝成一個jobGraph，并且是以jobGraph的形式提交。一個jobGraph是一個flink dataflow，它是由多個jobVertex組成的DAG。JobManager會將一個JobGraph轉(zhuǎn)換映射為一個ExecutionGraph
組件棧
Flink是一個分層架構(gòu)的系統(tǒng)，每一層所包含的組件都提供了特定的抽象，用來服務(wù)于上層組件，
- Deployment層，涉及了Flink的部署模式
啟動flink yarn session的時候，
- 最左邊的模塊
Flink YARN Clientcheck requested resources (containers and memory) are available，檢查資源可得性 - Client uploads a jar that contains Flink and the configuration to HDFS，上傳代碼和配置
- Client request a
YARN containerto start theApplicationMaster(AM，單個作業(yè)的資源管理和任務(wù)監(jiān)控模塊，以前是一個全局的JobTracker負(fù)責(zé)的，現(xiàn)在每個作業(yè)都一個)，啟動yarn AM - AM starts allocating the containers for Flink’s TaskManagers, which will download the jar file and the modified configuration from the HDFS
客戶端client負(fù)責(zé)向ResourceManager(RM)提交ApplicationMaster，并查詢應(yīng)用程序運(yùn)行狀態(tài)，ApplicationMaster(AM)負(fù)責(zé)向ResourceManager申請資源（以Container形式表示），并與NodeManager(NM)通信以啟動各個Container，此外，ApplicationMaster還負(fù)責(zé)監(jiān)控各個任務(wù)運(yùn)行狀態(tài)，并在失敗是為其重新申請資源。
flink RM Dispatcher，用于統(tǒng)一發(fā)布Job并監(jiān)控實(shí)例的運(yùn)行。但是可以選擇是否使用Dispatcher。
- Runtime層，提供了支持Flink計(jì)算的全部核心實(shí)現(xiàn)
- API層，實(shí)現(xiàn)了面向無界streaming的流處理和面向有界Batch的批處理接口
- Libraries層，F(xiàn)link應(yīng)用框架層，CEP復(fù)雜事件處理、Table基于SQL-like的關(guān)系操作、FlinkML機(jī)器學(xué)習(xí)、Gelly圖處理
內(nèi)部原理
容錯機(jī)制
Flink基于Checkpoint機(jī)制實(shí)現(xiàn)容錯，它的原理是不斷地生成分布式Streaming數(shù)據(jù)流Snapshot。在流處理失敗時，通過這些Snapshot可以恢復(fù)數(shù)據(jù)流處理。
checkpoint, snapshot, stream aligning, exactly once, at least once
調(diào)度機(jī)制
在jobManager，會接收到client提交的jobGraph形式的flink job，并將其轉(zhuǎn)換映射為executionGraph
- jobGraph是一個job的用戶邏輯視圖表示，將一個用戶要對數(shù)據(jù)流進(jìn)行的處理表示為單個DAG圖
- executionGraph是jobGraph的并行表示，也就是實(shí)際jobManager調(diào)度一個job在taskManager上運(yùn)行的邏輯視圖，也是一個DAG
上圖用戶提交的Flink Job對各個Operator進(jìn)行的配置（從下往上），即data source的并行度設(shè)置為4（最底層1個data source，但是其parallel=4），MapFunction的并行度也為4（中間層），ReduceFunction的并行度為3（頂層）。
迭代機(jī)制
機(jī)器學(xué)習(xí)和圖計(jì)算應(yīng)用，都會使用到迭代計(jì)算。flink通過迭代operator中定義step函數(shù)來實(shí)現(xiàn)迭代算法，包括Iterate和Delta Iterate兩類，
反壓機(jī)制
flink使用了高效有界的分布式阻塞隊(duì)列，就像java通用的blockingQueue。一個較慢的接收者會降低發(fā)送者的發(fā)送速率，因?yàn)橐坏┯薪珀?duì)列滿了發(fā)送者會被阻塞。
- 當(dāng)netty接收端發(fā)送數(shù)據(jù)時，為了將netty中的數(shù)據(jù)拷貝到task中（往task寫入數(shù)據(jù)），InputChannel會向其對應(yīng)的緩沖池localBufferPool申請內(nèi)存塊，
- 如果localBufferPool也沒有可用內(nèi)存塊且申請的數(shù)量還沒到池子（隊(duì)列）上限，則就向networkBufferPool申請內(nèi)存塊
- 如果localBufferPool已申請的數(shù)量達(dá)到上限了，或者networkBufferPool也沒有可用內(nèi)存塊，此時task的netty channel會暫停讀取，上游的發(fā)送端會立即響應(yīng)停止發(fā)送，拓?fù)溥M(jìn)入反壓狀態(tài)
- 當(dāng)task線程寫數(shù)據(jù)到resultPartition時（task數(shù)據(jù)往外寫），也會向池子請求內(nèi)存塊，如果沒有可用內(nèi)存塊時，也阻塞在請求內(nèi)存塊的地方，達(dá)到暫停寫入的目的
- 在一個內(nèi)存塊被消費(fèi)完成之后（在輸出端是指內(nèi)存塊中的字節(jié)寫入到netty channel；在輸入端是指內(nèi)存塊中的字節(jié)被反序列化成對象），會調(diào)用buffer.recycle()方法，將內(nèi)存塊還給localBufferPool，如果localBufferPool中當(dāng)前申請的數(shù)量超過了池子容量，則localBufferPool會將該內(nèi)存塊回收給networkBufferPool。如果沒超池子容量，則繼續(xù)留在localBufferPool中，減少反復(fù)申請的開銷
backPressure在流式計(jì)算系統(tǒng)中用于協(xié)調(diào)上、下游operator的處理速度。因?yàn)樵谝粋€stream上進(jìn)行處理的多個operator之間，它們的處理速度和方式可能非常不同，所以就存在上游operator如果處理速度過快，下游operator可能會堆積stream記錄。因此，對下游operator處理速度跟不上的情況，如果下游operator能夠?qū)⒆约禾幚頎顟B(tài)傳播給上游operator，使得上游operator處理速度慢下來，從而緩解上述問題。
JobManager會反復(fù)調(diào)用Task運(yùn)行所在線程的Thread.getStackTrace()，默認(rèn)情況下，JobManager會每隔50ms觸發(fā)對每個Task依次進(jìn)行100次堆棧跟蹤調(diào)用，根據(jù)調(diào)用調(diào)用結(jié)果來確定Backpressure，通過計(jì)算得到一個比值radio來確定當(dāng)前運(yùn)行Job的Backpressure狀態(tài)。在Web界面上可以看到這個Radio值，它表示在一個內(nèi)部方法調(diào)用中阻塞（Stuck）的堆棧跟蹤次數(shù)，例如，radio=0.01，表示100次中僅有1次方法調(diào)用阻塞。Flink目前定義了如下Backpressure狀態(tài)：
- OK: 0 <= Ratio <= 0.10
- LOW: 0.10 < Ratio <= 0.5
- HIGH: 0.5 < Ratio <= 1
