[TOC]
flink是什么
Flink主頁在其頂部展示了該項(xiàng)目的理念：“Apache Flink是為分布式、高性能、隨時(shí)可用以及準(zhǔn)確的流處理應(yīng)用程序打造的開源流處理框架”。Apache Flink是一個(gè)框架和分布式處理引擎，用于對***和有界數(shù)據(jù)流進(jìn)行有狀態(tài)計(jì)算。Flink被設(shè)計(jì)在所有常見的集群環(huán)境中運(yùn)行，以內(nèi)存執(zhí)行速度和任意規(guī)模來執(zhí)行計(jì)算。
flink基本框架
批處理的特點(diǎn)是有界、持久、大量，批處理非常適合需要訪問全套記錄才能完成的計(jì)算工作，一般用于離線統(tǒng)計(jì)。流處理的特點(diǎn)是***、實(shí)時(shí)，流處理方式無需針對整個(gè)數(shù)據(jù)集執(zhí)行操作，而是對通過系統(tǒng)傳輸?shù)拿總€(gè)數(shù)據(jù)項(xiàng)執(zhí)行操作，一般用于實(shí)時(shí)統(tǒng)計(jì)。
在Spark生態(tài)體系中，對于批處理和流處理采用了不同的技術(shù)框架，批處理由SparkSQL實(shí)現(xiàn)，流處理由Spark Streaming實(shí)現(xiàn)，這也是大部分框架采用的策略，使用獨(dú)立的處理器實(shí)現(xiàn)批處理和流處理，而Flink可以同時(shí)實(shí)現(xiàn)批處理和流處理。
Flink是如何同時(shí)實(shí)現(xiàn)批處理與流處理的呢？答案是，F(xiàn)link將批處理（即處理有限的靜態(tài)數(shù)據(jù)）視作一種特殊的流處理。
Flink的核心計(jì)算架構(gòu)是下圖中的Flink Runtime執(zhí)行引擎，它是一個(gè)分布式系統(tǒng)，能夠接受數(shù)據(jù)流程序并在一臺(tái)或多臺(tái)機(jī)器上以容錯(cuò)方式執(zhí)行。
Flink Runtime執(zhí)行引擎可以作為YARN（Yet Another Resource Negotiator）的應(yīng)用程序在集群上運(yùn)行，也可以在Mesos集群上運(yùn)行，還可以在單機(jī)上運(yùn)行（這對于調(diào)試Flink應(yīng)用程序來說非常有用）。
上圖為Flink技術(shù)棧的核心組成部分，值得一提的是，F(xiàn)link分別提供了面向流式處理的接口（DataStream API）和面向批處理的接口（DataSet API）。因此，F(xiàn)link既可以完成流處理，也可以完成批處理。Flink支持的拓展庫涉及機(jī)器學(xué)習(xí)（FlinkML）、復(fù)雜事件處理（CEP）、以及圖計(jì)算（Gelly），還有分別針對流處理和批處理的Table API。
能被Flink Runtime執(zhí)行引擎接受的程序很強(qiáng)大，但是這樣的程序有著冗長的代碼，編寫起來也很費(fèi)力，基于這個(gè)原因，F(xiàn)link提供了封裝在Runtime執(zhí)行引擎之上的API，以幫助用戶方便地生成流式計(jì)算程序。Flink 提供了用于流處理的DataStream API和用于批處理的DataSet API。值得注意的是，盡管Flink Runtime執(zhí)行引擎是基于流處理的，但是DataSet API先于DataStream API被開發(fā)出來，這是因?yàn)楣I(yè)界對無限流處理的需求在Flink誕生之初并不大。
DataStream API可以流暢地分析無限數(shù)據(jù)流，并且可以用Java或者Scala來實(shí)現(xiàn)。開發(fā)人員需要基于一個(gè)叫DataStream的數(shù)據(jù)結(jié)構(gòu)來開發(fā)，這個(gè)數(shù)據(jù)結(jié)構(gòu)用于表示永不停止的分布式數(shù)據(jù)流。
Flink的分布式特點(diǎn)體現(xiàn)在它能夠在成百上千臺(tái)機(jī)器上運(yùn)行，它將大型的計(jì)算任務(wù)分成許多小的部分，每個(gè)機(jī)器執(zhí)行一部分。Flink能夠自動(dòng)地確保發(fā)生機(jī)器故障或者其他錯(cuò)誤時(shí)計(jì)算能夠持續(xù)進(jìn)行，或者在修復(fù)bug或進(jìn)行版本升級后有計(jì)劃地再執(zhí)行一次。這種能力使得開發(fā)人員不需要擔(dān)心運(yùn)行失敗。Flink本質(zhì)上使用容錯(cuò)性數(shù)據(jù)流，這使得開發(fā)人員可以分析持續(xù)生成且永遠(yuǎn)不結(jié)束的數(shù)據(jù)（即流處理）。
1.4 無窮數(shù)據(jù)流和有限數(shù)據(jù)流
無窮數(shù)據(jù)集：無窮的持續(xù)集合的數(shù)據(jù)集合
有限數(shù)據(jù)集：有限不會(huì)改變的數(shù)據(jù)集合
常見的無窮數(shù)據(jù)集合有：
用戶與客戶端的實(shí)時(shí)交互數(shù)據(jù)
應(yīng)用實(shí)時(shí)產(chǎn)生的日志
金融市場的實(shí)時(shí)交易記錄
1.6 flink特性
1、高吞吐和低延遲性
2、支持 Event Time 和亂序事件
Flink 支持了流處理和 Event Time 語義的窗口機(jī)制。
Event time 使得計(jì)算亂序到達(dá)的事件或可能延遲到達(dá)的事件更加簡單。
3、狀態(tài)計(jì)算的 exactly-once 語義
故障狀態(tài)下，需要重啟計(jì)算任務(wù)，這時(shí)候需要避免已經(jīng)處理過的數(shù)據(jù)的重復(fù)處理。
流程序可以在計(jì)算過程中維護(hù)自定義狀態(tài)。
Flink 的 checkpointing 機(jī)制保證了即時(shí)在故障發(fā)生下也能保障狀態(tài)的 exactly once 語義。
4、高度靈活的流式窗口
Flink 支持在時(shí)間窗口，統(tǒng)計(jì)窗口，session 窗口，以及數(shù)據(jù)驅(qū)動(dòng)的窗口
窗口可以通過靈活的觸發(fā)條件來定制，以支持復(fù)雜的流計(jì)算模式。
5、帶反壓的連續(xù)流模型
數(shù)據(jù)流應(yīng)用執(zhí)行的是不間斷的（常駐）operators。
Flink streaming 在運(yùn)行時(shí)有著天然的流控：慢的數(shù)據(jù) sink 節(jié)點(diǎn)會(huì)反壓（backpressure）快的數(shù)據(jù)源（sources）。
6、容錯(cuò)性
Flink 的容錯(cuò)機(jī)制是基于 Chandy-Lamport distributed snapshots 來實(shí)現(xiàn)的。
這種機(jī)制是非常輕量級的，允許系統(tǒng)擁有高吞吐率的同時(shí)還能提供強(qiáng)一致性的保障。
7、Batch 和 Streaming 一個(gè)系統(tǒng)流處理和批處理共用一個(gè)引擎
Flink 為流處理和批處理應(yīng)用公用一個(gè)通用的引擎。批處理應(yīng)用可以以一種特殊的流處理應(yīng)用高效地運(yùn)行。
8、內(nèi)存管理
Flink 在 JVM 中實(shí)現(xiàn)了自己的內(nèi)存管理。
應(yīng)用可以超出主內(nèi)存的大小限制，并且承受更少的垃圾收集的開銷。
二、Flink基本架構(gòu)
2.1 flink中的角色
Flink運(yùn)行時(shí)包含了兩種類型的處理器：
JobManager處理器：也稱之為Master，用于協(xié)調(diào)分布式執(zhí)行，它們用來調(diào)度task，協(xié)調(diào)檢查點(diǎn)，協(xié)調(diào)失敗時(shí)恢復(fù)等。Flink運(yùn)行時(shí)至少存在一個(gè)master處理器，如果配置高可用模式則會(huì)存在多個(gè)master處理器，它們其中有一個(gè)是leader，而其他的都是standby。
TaskManager處理器：也稱之為Worker，用于執(zhí)行一個(gè)dataflow的task(或者特殊的subtask)、數(shù)據(jù)緩沖和data stream的交換，F(xiàn)link運(yùn)行時(shí)至少會(huì)存在一個(gè)worker處理器。
Master和Worker處理器可以直接在物理機(jī)上啟動(dòng)，或者通過像YARN這樣的資源調(diào)度框架。Worker連接到Master，告知自身的可用性進(jìn)而獲得任務(wù)分配。
2.2 ***數(shù)據(jù)流與有界數(shù)據(jù)流
數(shù)據(jù)流：
數(shù)據(jù)流有一個(gè)開始但是沒有結(jié)束，它們不會(huì)在生成時(shí)終止并提供數(shù)據(jù)，必須連續(xù)處理流，也就是說必須在獲取后立即處理event。對于數(shù)據(jù)流我們無法等待所有數(shù)據(jù)都到達(dá)，因?yàn)檩斎胧?strong>的，并且在任何時(shí)間點(diǎn)都不會(huì)完成。處理數(shù)據(jù)通常要求以特定順序（例如事件發(fā)生的順序）獲取event，以便能夠推斷結(jié)果完整性。
有界數(shù)據(jù)流：
有界數(shù)據(jù)流有明確定義的開始和結(jié)束，可以在執(zhí)行任何計(jì)算之前通過獲取所有數(shù)據(jù)來處理有界流，處理有界流不需要有序獲取，因?yàn)榭梢允冀K對有界數(shù)據(jù)集進(jìn)行排序，有界流的處理也稱為批處理。
Apache Flink是一個(gè)面向分布式數(shù)據(jù)流處理和批量數(shù)據(jù)處理的開源計(jì)算平臺(tái)，它能夠基于同一個(gè)Flink運(yùn)行時(shí)(Flink Runtime)，提供支持流處理和批處理兩種類型應(yīng)用的功能?，F(xiàn)有的開源計(jì)算方案，會(huì)把流處理和批處理作為兩種不同的應(yīng)用類型，因?yàn)樗鼈円獙?shí)現(xiàn)的目標(biāo)是完全不相同的：流處理一般需要支持低延遲、Exactly-once保證，而批處理需要支持高吞吐、高效處理，所以在實(shí)現(xiàn)的時(shí)候通常是分別給出兩套實(shí)現(xiàn)方法，或者通過一個(gè)獨(dú)立的開源框架來實(shí)現(xiàn)其中每一種處理方案。例如，實(shí)現(xiàn)批處理的開源方案有MapReduce、Tez、Crunch、Spark，實(shí)現(xiàn)流處理的開源方案有Samza、Storm。
Flink在實(shí)現(xiàn)流處理和批處理時(shí)，與傳統(tǒng)的一些方案完全不同，它從另一個(gè)視角看待流處理和批處理，將二者統(tǒng)一起來：Flink是完全支持流處理，也就是說作為流處理看待時(shí)輸入數(shù)據(jù)流是***的；批處理被作為一種特殊的流處理，只是它的輸入數(shù)據(jù)流被定義為有界的?；谕粋€(gè)Flink運(yùn)行時(shí)(Flink Runtime)，分別提供了流處理和批處理API，而這兩種API也是實(shí)現(xiàn)上層面向流處理、批處理類型應(yīng)用框架的基礎(chǔ)。
2.3 flink數(shù)據(jù)流編程接口抽象
Flink提供了不同級別的抽象，以開發(fā)流或批處理作業(yè)，如下圖所示：
最底層級的抽象僅僅提供了有狀態(tài)流，它將通過過程函數(shù)（Process Function）被嵌入到DataStream API中。底層過程函數(shù)（Process Function） 與 DataStream API 相集成，使其可以對某些特定的操作進(jìn)行底層的抽象，它允許用戶可以自由地處理來自一個(gè)或多個(gè)數(shù)據(jù)流的事件，并使用一致的容錯(cuò)的狀態(tài)。除此之外，用戶可以注冊事件時(shí)間并處理時(shí)間回調(diào)，從而使程序可以處理復(fù)雜的計(jì)算。
實(shí)際上，大多數(shù)應(yīng)用并不需要上述的底層抽象，而是針對核心API（Core APIs） 進(jìn)行編程，比如DataStream API（有界或***流數(shù)據(jù)）以及DataSet API（有界數(shù)據(jù)集）。這些API為數(shù)據(jù)處理提供了通用的構(gòu)建模塊，比如由用戶定義的多種形式的轉(zhuǎn)換（transformations），連接（joins），聚合（aggregations），窗口操作（windows）等等。DataSet API 為有界數(shù)據(jù)集提供了額外的支持，例如循環(huán)與迭代。這些API處理的數(shù)據(jù)類型以類（classes）的形式由各自的編程語言所表示。
Table API 是以表為中心的聲明式編程，其中表可能會(huì)動(dòng)態(tài)變化（在表達(dá)流數(shù)據(jù)時(shí)）。Table API遵循（擴(kuò)展的）關(guān)系模型：表有二維數(shù)據(jù)結(jié)構(gòu)（schema）（類似于關(guān)系數(shù)據(jù)庫中的表），同時(shí)API提供可比較的操作，例如select、project、join、group-by、aggregate等。Table API程序聲明式地定義了什么邏輯操作應(yīng)該執(zhí)行，而不是準(zhǔn)確地確定這些操作代碼的看上去如何 。 盡管Table API可以通過多種類型的用戶自定義函數(shù)（UDF）進(jìn)行擴(kuò)展，其仍不如核心API更具表達(dá)能力，但是使用起來卻更加簡潔（代碼量更少）。除此之外，Table API程序在執(zhí)行之前會(huì)經(jīng)過內(nèi)置優(yōu)化器進(jìn)行優(yōu)化。
你可以在表與 DataStream/DataSet 之間無縫切換，以允許程序?qū)?Table API 與 DataStream 以及 DataSet 混合使用。
Flink提供的最高層級的抽象是 SQL 。這一層抽象在語法與表達(dá)能力上與 Table API 類似，但是是以SQL查詢表達(dá)式的形式表現(xiàn)程序。SQL抽象與Table API交互密切，同時(shí)SQL查詢可以直接在Table API定義的表上執(zhí)行。
三、flink運(yùn)行架構(gòu)
3.1 提交任務(wù)到y(tǒng)arn的流程
flink在生產(chǎn)中，一般是使用yarn作為資源調(diào)度平臺(tái)，比較少使用standalone的方式進(jìn)行資源調(diào)度。所以這里以yarn為例，說明flink提交任務(wù)到y(tǒng)arn的流程。
Flink任務(wù)提交后，Client向HDFS上傳Flink的Jar包和配置，之后向Yarn ResourceManager提交任務(wù)，ResourceManager分配Container資源并通知對應(yīng)的NodeManager啟動(dòng)ApplicationMaster，ApplicationMaster啟動(dòng)后加載Flink的Jar包和配置構(gòu)建環(huán)境，然后啟動(dòng)JobManager，之后ApplicationMaster向ResourceManager申請資源啟動(dòng)TaskManager，ResourceManager分配Container資源后，由ApplicationMaster通知資源所在節(jié)點(diǎn)的NodeManager啟動(dòng)TaskManager，NodeManager加載Flink的Jar包和配置構(gòu)建環(huán)境并啟動(dòng)TaskManager，TaskManager啟動(dòng)后向JobManager發(fā)送心跳包，并等待JobManager向其分配任務(wù)。
3.2 任務(wù)調(diào)度組件
1、 Program Code：我們編寫的 Flink 應(yīng)用程序代碼
2、 Job Client：Job Client 不是 Flink 程序執(zhí)行的內(nèi)部部分，但它是任務(wù)執(zhí)行的起點(diǎn)。 Job Client 負(fù)責(zé)接受用戶的程序代碼，然后創(chuàng)建數(shù)據(jù)流，將數(shù)據(jù)流提交給 Job Manager 以便進(jìn)一步執(zhí)行。 執(zhí)行完成后，Job Client 將結(jié)果返回給用戶
- 3、 JobManager：主進(jìn)程（也稱為作業(yè)管理器）協(xié)調(diào)和管理程序的執(zhí)行。 它的主要職責(zé)包括安排任務(wù)，管理checkpoint ，故障恢復(fù)等。機(jī)器集群中至少要有一個(gè) master，master 負(fù)責(zé)調(diào)度 task，協(xié)調(diào) checkpoints 和容災(zāi)，高可用設(shè)置的話可以有多個(gè) master，但要保證一個(gè)是active, 其他是 standby; Job Manager 包含 Actor system(通信系統(tǒng))、Scheduler（調(diào)度）、Check pointing 三個(gè)重要的組件
4、 Task Manager：從 Job Manager 處接收需要部署的 Task。Task Manager 是在 JVM 中的一個(gè)或多個(gè)線程中執(zhí)行任務(wù)的工作節(jié)點(diǎn)。 任務(wù)執(zhí)行的并行性由每個(gè) Task Manager 上可用的任務(wù)槽（task slot）決定。 每個(gè)任務(wù)代表分配給任務(wù)槽的一組資源。 例如，如果 Task Manager 有四個(gè)插槽，那么它將為每個(gè)插槽分配 25％ 的內(nèi)存。 可以在任務(wù)槽中運(yùn)行一個(gè)或多個(gè)線程。 同一插槽中的線程共享相同的 JVM。 同一 JVM 中的任務(wù)共享 TCP 連接和心跳消息。Task Manager 的一個(gè) Slot 代表一個(gè)可用線程，該線程具有固定的內(nèi)存，注意 Slot 只對內(nèi)存隔離，沒有對 CPU 隔離。默認(rèn)情況下，F(xiàn)link 允許子任務(wù)共享 Slot，即使它們是不同 task 的 subtask，只要它們來自相同的 job。這種共享可以有更好的資源利用率。
3.3 TaskManager和slots原理
每一個(gè)worker(TaskManager)是一個(gè)JVM進(jìn)程，它可能會(huì)在獨(dú)立的線程上執(zhí)行一個(gè)或多個(gè)subtask。為了控制一個(gè)worker能接收多少個(gè)task，worker通過task slot來進(jìn)行控制（一個(gè)worker至少有一個(gè)task slot）。
每個(gè)task slot表示TaskManager擁有資源的一個(gè)固定大小的子集。假如一個(gè)TaskManager有三個(gè)slot，那么它會(huì)將其管理的內(nèi)存平均分成三份給各個(gè)slot。資源slot化意味著一個(gè)subtask將不需要跟來自其他job的subtask競爭被管理的內(nèi)存，取而代之的是它將擁有一定數(shù)量的內(nèi)存儲(chǔ)備。需要注意的是，這里不會(huì)涉及到CPU的隔離，slot目前僅僅用來隔離task的受管理的內(nèi)存。
通過調(diào)整task slot的數(shù)量，允許用戶定義subtask之間如何互相隔離。如果一個(gè)TaskManager一個(gè)slot，那將意味著每個(gè)task group運(yùn)行在獨(dú)立的JVM中（該JVM可能是通過一個(gè)特定的容器啟動(dòng)的），而一個(gè)TaskManager多個(gè)slot意味著更多的subtask可以共享同一個(gè)JVM。而在同一個(gè)JVM進(jìn)程中的task將共享TCP連接（基于多路復(fù)用）和心跳消息。它們也可能共享數(shù)據(jù)集和數(shù)據(jù)結(jié)構(gòu)，因此這減少了每個(gè)task的負(fù)載。
Task Slot是靜態(tài)的概念，是指TaskManager具有的并發(fā)執(zhí)行能力，可以通過參數(shù)taskmanager.numberOfTaskSlots進(jìn)行配置，而并行度parallelism是動(dòng)態(tài)概念，即TaskManager運(yùn)行程序時(shí)實(shí)際使用的并發(fā)能力，可以通過參數(shù)parallelism.default進(jìn)行配置。
也就是說，假設(shè)一共有3個(gè)TaskManager，每一個(gè)TaskManager中的分配3個(gè)TaskSlot，也就是每個(gè)TaskManager可以接收3個(gè)task，一共9個(gè)TaskSlot，如果我們設(shè)置parallelism.default=1，即運(yùn)行程序默認(rèn)的并行度為1，9個(gè)TaskSlot只用了1個(gè)，有8個(gè)空閑，因此，設(shè)置合適的并行度才能提高效率。實(shí)際上slots限制的限制了該taskmanager在整個(gè)集群中能夠并行運(yùn)行task的數(shù)目，而parallelism.default則是限制單個(gè)job能夠使用slot的數(shù)量，但是允許多個(gè)job同時(shí)運(yùn)行，所以實(shí)際上是對單個(gè)job的并發(fā)限制。
3.4 程序與數(shù)據(jù)流
Flink程序的基礎(chǔ)構(gòu)建模塊是 流（streams） 與 轉(zhuǎn)換（transformations）（需要注意的是，F(xiàn)link的DataSet API所使用的DataSets其內(nèi)部也是stream）。一個(gè)stream可以看成一個(gè)中間結(jié)果，而一個(gè)transformations是以一個(gè)或多個(gè)stream作為輸入的某種operation，該operation利用這些stream進(jìn)行計(jì)算從而產(chǎn)生一個(gè)或多個(gè)result stream。
在運(yùn)行時(shí)，F(xiàn)link上運(yùn)行的程序會(huì)被映射成streaming dataflows，它包含了streams和transformations operators。每一個(gè)dataflow以一個(gè)或多個(gè)sources開始以一個(gè)或多個(gè)sinks結(jié)束。dataflow類似于任意的有向無環(huán)圖（DAG），當(dāng)然特定形式的環(huán)可以通過iteration構(gòu)建。在大部分情況下，程序中的transformations跟dataflow中的operator是一一對應(yīng)的關(guān)系，但有時(shí)候，一個(gè)transformation可能對應(yīng)多個(gè)operator。
