Apache Flink 官方文檔--概念

數(shù)據(jù)流編程模型

原文鏈接
博主理解篇

抽象層次

??Flink提供不同級(jí)別的抽象來開發(fā)流/批處理應(yīng)用程序。

• 這個(gè)最低級(jí)別的抽象提供了有狀態(tài)的流式操作。它是通過處理函數(shù)嵌入到DataStream API。它允許用戶自由的處理一個(gè)或者多個(gè)數(shù)據(jù)流中的事件，并且使用一致，容錯(cuò)的狀態(tài)。此外，用戶可以注冊(cè)事件時(shí)間和處理時(shí)間回調(diào)，允許程序?qū)崿F(xiàn)復(fù)雜的計(jì)算。
• 實(shí)際上，大多數(shù)應(yīng)用不需要上面描述的低級(jí)別抽象，而是針對(duì)Core APIs(核心API)，例如：?DataStream API(有邊界和無邊界的數(shù)據(jù)流) 和?DataSet API(有邊界的數(shù)據(jù)集)。這些流暢的API提供通用數(shù)據(jù)處理，像用戶指定的各種形式的transformations(轉(zhuǎn)換)，joins(連接)，aggregations(聚合)，windows(窗口化操作)，state(狀態(tài))等等。這些API表示在各自的編程語(yǔ)言中為類(class)中的數(shù)據(jù)類型進(jìn)行處理。

??低階的處理函數(shù)集成了DataStream API，這樣就可以針對(duì)特性的操作使用低層級(jí)的抽象。DataSet API 為有邊界的 data sets提供了附加的原語(yǔ)，例如循環(huán)/迭代。

• Table API 是一種以表為中心的聲明式的DSL，它可能會(huì)被動(dòng)態(tài)的改變(當(dāng)處理數(shù)據(jù)流的時(shí)候)。Table API?遵循擴(kuò)展模型：Table 有一個(gè)附加模式(類似于關(guān)系型數(shù)據(jù)庫(kù)表)并且API提供了類似的操作，例如：select，project，join，group-by，aggregate等等。Table API聲明式的定義了邏輯操作應(yīng)該怎么做 而不是確切的指定操作的代碼看起來如何。盡管table API可以通過多種形式的用戶自定義函數(shù)來擴(kuò)展，它的表現(xiàn)還是不如Core APIs，但是用起來更加的簡(jiǎn)潔(寫更少的代碼)。此外，Table API 還可以執(zhí)行一個(gè)優(yōu)化器，適用于優(yōu)化規(guī)則之前執(zhí)行。
  ??Table和DataStream/DataSet之間可以無縫的轉(zhuǎn)換，允許程序組合使用Table api和DataStream和DataSet的API。
• Flink最高級(jí)別的抽象是sql。這種抽象在語(yǔ)義和表達(dá)上面類似于Table API，但將程序表示為SQL查詢表達(dá)式。SQL抽象與Table API 緊密聯(lián)系在一起，Sql查詢可以在table API定義的表中執(zhí)行。

  程序和數(shù)據(jù)流

  ??Flink程序的基本構(gòu)建模塊是streams(流)和transformations(轉(zhuǎn)換)。(需要注意的是，F(xiàn)link的DataSet API所使用的DataSets內(nèi)部也是流-更多內(nèi)容將在以后解釋)。從概念上講流(可能沒有結(jié)束)是一個(gè)數(shù)據(jù)流記錄，而轉(zhuǎn)換是一個(gè)操作，它取一種或者多個(gè)流作為輸入，并產(chǎn)生一個(gè)或者多個(gè)輸出流作為結(jié)果。
  ??當(dāng)執(zhí)行的時(shí)候，F(xiàn)link程序映射到streaming dataflows(流數(shù)據(jù)流)，由streams和轉(zhuǎn)換operators組成。每一個(gè)數(shù)據(jù)流開始于一個(gè)或者多個(gè)source，并且終止于一個(gè)或者多個(gè)sink。數(shù)據(jù)流類似于任意的有向無環(huán)圖(DAGS)。雖然通過迭代構(gòu)造允許特定形式的環(huán)，但是大多數(shù)情況下，簡(jiǎn)單起見，我們都不考慮這一點(diǎn)。
  
  ??通常情況下，程序中的轉(zhuǎn)換與數(shù)據(jù)流中的操作是一一對(duì)應(yīng)的。有時(shí)，然而，一個(gè)轉(zhuǎn)換可能有多個(gè)轉(zhuǎn)換操作構(gòu)成。
  ??source和sink的文檔在streaming connectors和 batch connectors 。Transformation的文檔在DataStream operators和DataSet transformation。

  并行數(shù)據(jù)流

  ??Flink程序本質(zhì)上是并行的和分布式的。在執(zhí)行過程中，一個(gè)流(stream)包含一個(gè)或多個(gè)流分區(qū) (stream partition)，而每一個(gè)operator包含一個(gè)或多個(gè)operator子任務(wù) 。操作子任務(wù)之間彼此獨(dú)立，在不同的線程中執(zhí)行，甚至有可能運(yùn)行在不同的機(jī)器或容器上。
  ??operator子任務(wù)的數(shù)量即是此特定operator的并行度 。一個(gè)流的并行度即其生產(chǎn)operator的并行度。相同程序中的不同的operator可能有不同級(jí)別的并行度。
  
  ??流在兩個(gè)operator之間傳輸數(shù)據(jù)，可以通過一對(duì)一（或稱 forwarding ）模式，或者通過redistributing模式：

• 一對(duì)一流（例如上圖中Source與map() opreator之間）保持了元素的分區(qū)與排序。那意味著 map() operator的子任務(wù)[1]將以與 Source 的子任務(wù)[1]生成順序相同的順序查看到相同的元素。
• Redistributing 流（如上圖中 map() 與 keyBy/window 之間，以及 keyBy/window 與 Sink 之間）則改變了流的分區(qū)。每一個(gè)operator子任務(wù)根據(jù)所選擇的轉(zhuǎn)換，向不同的目標(biāo)子任務(wù)發(fā)送數(shù)據(jù)。比如 keyBy() （根據(jù)key的哈希值重新分區(qū)）， broadcast() ，或者 rebalance() （隨機(jī)重分區(qū)）。在一次 redistributing 交換中，元素間的排序只保留在每對(duì)發(fā)送與接受子任務(wù)中（比如， map() 的子任務(wù)[1]與 keyBy/window 的子任務(wù)[2]）。因此在這個(gè)例子中，每個(gè)鍵的順序被保留下來，但是并行確實(shí)引入了不確定性--對(duì)于不同鍵的聚合結(jié)果到達(dá)sink的順序。
  ??配置和并行度的詳細(xì)配置可以查看這個(gè)文檔parallel execution。

  窗口（Window）

  ??聚合事件（比如計(jì)數(shù)、求和）在流上的工作方式與批處理不同。比如，對(duì)流中的所有元素進(jìn)行計(jì)數(shù)是不可能的，因?yàn)橥ǔＡ魇菬o限的（無邊界的）。相反，流上的聚合需要由窗口來劃定范圍，比如 “計(jì)算過去的5分鐘” ，或者 “最后100個(gè)元素的和” 。
  ??窗口可以是事件驅(qū)動(dòng)的 （比如：每30秒）或者數(shù)據(jù)驅(qū)動(dòng)的 （比如：每100個(gè)元素）。窗口通常被區(qū)分為不同的類型，比如滾動(dòng)窗口 （沒有重疊）， 滑動(dòng)窗口 （有重疊），以及會(huì)話窗口（由不活動(dòng)的間隙所打斷）
  
  ??更多的窗口例子可以查看這個(gè)博客。更多的明細(xì)可以查看窗口文檔window docs。

時(shí)間（Time）

??當(dāng)提到流程序（例如定義窗口）中的時(shí)間時(shí)，你可以參考不同的時(shí)間概念：

• 事件時(shí)間是事件創(chuàng)建的時(shí)間。它通常由事件中的時(shí)間戳描述，例如附接在生產(chǎn)傳感器，或者生產(chǎn)服務(wù)。Flink通過時(shí)間戳分配器訪問事件時(shí)間戳。
• 攝入時(shí)間是事件進(jìn)入Flink數(shù)據(jù)流源算子的時(shí)間。
• 處理時(shí)間 是每一個(gè)執(zhí)行時(shí)間操作的operator的本地時(shí)間。
  
  操作時(shí)間的更多詳細(xì)信息請(qǐng)查看文檔event time docs。

有狀態(tài)的操作

??盡管數(shù)據(jù)流中的很多操作一次只查看一個(gè)獨(dú)立的事件（比如事件解析器），有些操作卻會(huì)記錄多個(gè)事件間的信息（比如窗口算子）。 這些操作被稱為有狀態(tài)的 。
??有狀態(tài)操作的狀態(tài)保存在一個(gè)可被視作嵌入式鍵/值存儲(chǔ)的部分中。狀態(tài)由有狀態(tài)operator讀取的流一起被嚴(yán)格地分區(qū)與分布。因此，只能訪問一個(gè) keyBy() 函數(shù)之后的 keyed streams 的鍵/值狀態(tài)，并且僅限于與當(dāng)前事件鍵相關(guān)聯(lián)的值。調(diào)整流和狀態(tài)的鍵確保了所有狀態(tài)更新都是本地操作，以在沒有事務(wù)開銷的情況下確保一致性。這種對(duì)齊還使得Flink可以透明地重新分配狀態(tài)與調(diào)整流的分區(qū)。

查看更多信息，請(qǐng)查看此文檔有關(guān)state的內(nèi)容。

容錯(cuò)檢查點(diǎn)

??Flink使用流重放與檢查點(diǎn)的結(jié)合實(shí)現(xiàn)了容錯(cuò)。檢查點(diǎn)與每個(gè)輸入流的特定點(diǎn)及與相關(guān)的每一個(gè)operator的狀態(tài)相關(guān)。一個(gè)數(shù)據(jù)流可以從一個(gè)檢查點(diǎn)恢復(fù)出來，其中通過恢復(fù)operator狀態(tài)并從檢查點(diǎn)重放事件以保持一致性 （一次處理語(yǔ)義）
??檢查點(diǎn)間隔是以恢復(fù)時(shí)間（需要重放的事件數(shù)量）來消除執(zhí)行過程中容錯(cuò)的開銷的一種手段。
??容錯(cuò)內(nèi)部的描述提供了更多關(guān)于flink管理檢查點(diǎn)和相關(guān)的話題。啟用和配置檢查點(diǎn)的詳細(xì)信息請(qǐng)查看這個(gè)文檔checkpointing API docs。

流式批處理

??Flink將批處理程序作為流處理程序的特殊情況來執(zhí)行，只是流是有界的（有限個(gè)元素）。 DataSet 內(nèi)部被視為數(shù)據(jù)流。上述適用于流處理程序的概念同樣適用于批處理程序，除了一些例外：

• 批處理程序的容錯(cuò)不再使用檢查點(diǎn)。而是通過完全地重放流來恢復(fù)。因?yàn)檩斎胧怯薪绲?，因此這是可行的。這種方法使得恢復(fù)的成本增加，但是由于避免了檢查點(diǎn)，因而使得正常處理的開銷更小。
• DataSet API中的有狀態(tài)操作使用簡(jiǎn)化的in-memory/out-of-core數(shù)據(jù)結(jié)構(gòu)，而不是鍵/值索引。
• DataSet API引入了特殊的同步（superstep-base）迭代，而這種迭代僅僅能在有界流上執(zhí)行。細(xì)節(jié)可以查看迭代文檔。

分布式運(yùn)行時(shí)

原文鏈接

任務(wù)和Operator鏈

??對(duì)于分布式運(yùn)行，F(xiàn)link將operator子任務(wù)鏈接在一起放入任務(wù)池。每個(gè)任務(wù)由一個(gè)線程執(zhí)行。將operator鏈接到任務(wù)池中是一項(xiàng)有用的優(yōu)化：它減少線程到線程的切換和緩沖的開銷，并在降低延遲的同時(shí)提高整體吞吐量?？梢耘渲面溄有袨?，有關(guān)詳細(xì)信息，請(qǐng)查閱鏈接文檔。
??下圖中的示例數(shù)據(jù)流由五個(gè)子任務(wù)執(zhí)行，因此有五個(gè)并行線程。

作業(yè)管理器，任務(wù)管理器，客戶端

??Flink運(yùn)行時(shí)有兩種類型的進(jìn)程組成：

• 作業(yè)管理器(JobManagers,也稱為主節(jié)點(diǎn)master)負(fù)責(zé)協(xié)調(diào)分布式運(yùn)行時(shí)。它們調(diào)度任務(wù)，協(xié)調(diào)檢查點(diǎn)，協(xié)調(diào)失敗恢復(fù)，等。
  ??至少有一個(gè)作業(yè)管理器節(jié)點(diǎn)，高可用的環(huán)境有多個(gè)作業(yè)管理器，其中一個(gè)節(jié)點(diǎn)是leader角色，其他節(jié)點(diǎn)是standby角色。
• 任務(wù)管理器(TaskManagers,也稱為工作節(jié)點(diǎn)worker)執(zhí)行數(shù)據(jù)流的任務(wù)（更特定一些，子任務(wù)），緩沖以及轉(zhuǎn)換數(shù)據(jù)流。
  ??同樣至少有一個(gè)任務(wù)管理器節(jié)點(diǎn)。
  ??作業(yè)管理器與任務(wù)管理器可以以多種方式啟動(dòng)：以standalone集群的方式直接在主機(jī)上啟動(dòng)，或者被資源管理器YARN或Mesos管理啟動(dòng)。任務(wù)管理器連接到作業(yè)管理器上，聲明它們自己是可用狀態(tài)并且可被分配任務(wù)。
  ??客戶端不是程序運(yùn)行時(shí)的一部分，但是經(jīng)常用來準(zhǔn)備以及發(fā)送數(shù)據(jù)流程序到作業(yè)管理器上。此后，客戶端可以斷開連接，或者保持連接結(jié)束進(jìn)程報(bào)告?？蛻舳丝梢砸訨ava/Scala程序的方式觸發(fā)執(zhí)行，或者在命令行"./bin/flink"中運(yùn)行。
  

  Task Slot和資源

  ??每個(gè)Worker節(jié)點(diǎn)（任務(wù)管理器）是一個(gè)JVM進(jìn)程，在分開的線程中可以執(zhí)行一個(gè)或多個(gè)子任務(wù)。一個(gè)Worker通過控制task slots（至少一個(gè)）來控制節(jié)點(diǎn)接受多少任務(wù)。
  ??每個(gè)task slot代表任務(wù)管理器固定大小的資源子集。例如：一個(gè)擁有3個(gè)slot的任務(wù)管理器，將會(huì)分配它管理的1/3內(nèi)存到每個(gè)slot。對(duì)資源進(jìn)行分槽(slot)意味著子任務(wù)不會(huì)與其他作業(yè)的子任務(wù)競(jìng)爭(zhēng)管理的內(nèi)存，而是具有一定數(shù)量的保留管理內(nèi)存。注意此處沒有CPU隔離發(fā)生,現(xiàn)在只能分隔任務(wù)的管理內(nèi)存。
  ?? 通過調(diào)整task slot數(shù)目，用戶可以定義子任務(wù)之間如何隔離。每個(gè)任務(wù)管理器擁有一個(gè)slot意味著任務(wù)組運(yùn)行在隔離的JVM（例如，可以在隔離的容器上啟動(dòng)）上。擁有多個(gè)slots意味著更多的子任務(wù)共享相同的JVM。在相同的JVM上的任務(wù)共享TCP連接（通過多路復(fù)用）和心跳信息。它們還可以共享數(shù)據(jù)集和數(shù)據(jù)結(jié)構(gòu)，從而減少每個(gè)任務(wù)的開銷。
  
  ?? 默認(rèn)情況下，F(xiàn)link允許子任務(wù)共享slot，即使它們是不同任務(wù)的子任務(wù)，只要它們來自同一個(gè)作業(yè)。結(jié)果就是一個(gè)slot擁有這個(gè)作業(yè)的所有管道操作（pipeline）。允許這種slot共享有兩個(gè)主要的好處：

• Flink集群需要與作業(yè)中使用的最高并行度同樣的task slots。無需計(jì)算程序總共包含多少任務(wù)（在不同的并行度之上）。
• 更容易獲得更好的資源利用率。沒有共享slot的情況下，非密集型的source與map()子任務(wù)將會(huì)阻塞與資源密集型的窗口子任務(wù)同樣多的資源。通過共享slot，將并行度從2增加到6可以充分利用the slotted（時(shí)隙）資源，同時(shí)確保繁重的子任務(wù)在任務(wù)管理器上公平的分配。
  
  ??APIs同樣還包括用于防止不期望的slot共享的資源組機(jī)制。
  ??根據(jù)經(jīng)驗(yàn)，一個(gè)很好的默認(rèn)任務(wù)槽(task slot)數(shù)就是CPU核心數(shù)。使用超線程，每個(gè)slot需要2個(gè)或更多硬件線程上下文。

狀態(tài)后端(State Backends)

??存儲(chǔ)鍵/值對(duì)索引的確切數(shù)據(jù)結(jié)構(gòu)取決于所選的狀態(tài)后端。一個(gè)狀態(tài)后端將數(shù)據(jù)存儲(chǔ)在內(nèi)存中的哈希映射中，另一個(gè)狀態(tài)后端使用[RocksDB]()作為鍵/值對(duì)存儲(chǔ)。除了定義保存狀態(tài)的數(shù)據(jù)結(jié)構(gòu)之外，狀態(tài)后端還實(shí)現(xiàn)邏輯以獲取鍵/值對(duì)狀態(tài)的時(shí)間點(diǎn)快照，并將該快照存儲(chǔ)為檢查點(diǎn)的一部分。

保存點(diǎn)Savepoints

??用Data Stream API編寫的程序可以從保存點(diǎn)恢復(fù)執(zhí)行。保存點(diǎn)允許更新程序和Flink集群，而不會(huì)丟失任何狀態(tài)。
??保存點(diǎn)是手動(dòng)觸發(fā)的檢查點(diǎn)，它會(huì)獲取程序的快照并將其寫入狀態(tài)后端。他們依靠常規(guī)的檢查點(diǎn)機(jī)制。在執(zhí)行期間，程序會(huì)定期在工作節(jié)點(diǎn)上創(chuàng)建快照并生成檢查點(diǎn)。對(duì)于恢復(fù)，僅僅需要最后完成的檢查點(diǎn)，因此一旦新的檢查點(diǎn)完成，就可以安全地丟棄舊的檢查點(diǎn)。
??保存點(diǎn)與這些定期檢查點(diǎn)類似，不同之處在于它們由用戶觸發(fā)，并且在較新的檢查點(diǎn)完成時(shí)不會(huì)自動(dòng)過期。可以從命令行或通過REST API取消作業(yè)時(shí)創(chuàng)建保存點(diǎn)。