本文目錄:

一、Flink簡介

二、Flink 部署及啟動

三、Flink 運行架構

四、Flink 算子大全

五、流處理中的 Time 與 Window

六、Flink 狀態(tài)管理

七、Flink 容錯

八、Flink SQL

九、Flink CEP

十、Flink CDC

十一、基于 Flink 構建全場景實時數(shù)倉

十二、Flink 大廠面試題

Flink 涉及的知識點如下圖所示,本文將逐一講解:

本文檔參考了 Flink 的官網(wǎng)及其他眾多資料整理而成,為了整潔的排版及舒適的閱讀,對于模糊不清晰的圖片及黑白圖片進行重新繪制成了高清彩圖。

一、Flink 簡介1． Flink 發(fā)展

這幾年大數(shù)據(jù)的飛速發(fā)展,出現(xiàn)了很多熱門的開源社區(qū),其中著名的有 Hadoop、Storm,以及后來的 Spark,他們都有著各自專注的應用場景。Spark 掀開了內(nèi)存計算的先河,也以內(nèi)存為賭注,贏得了內(nèi)存計算的飛速發(fā)展。Spark 的火熱或多或少的掩蓋了其他分布式計算的系統(tǒng)身影。就像 Flink,也就在這個時候默默的發(fā)展著。

在國外一些社區(qū),有很多人將大數(shù)據(jù)的計算引擎分成了 4 代,當然,也有很多人不會認同。我們先姑且這么認為和討論。

首先第一代的計算引擎,無疑就是 Hadoop 承載的 MapReduce。這里大家應該都不會對 MapReduce 陌生,它將計算分為兩個階段,分別為 Map 和 Reduce。對于上層應用來說,就不得不想方設法去拆分算法,甚至于不得不在上層應用實現(xiàn)多個 Job 的串聯(lián),以完成一個完整的算法,例如迭代計算。

由于這樣的弊端,催生了支持 DAG 框架的產(chǎn)生。因此,支持 DAG 的框架被劃分為第二代計算引擎。如 Tez 以及更上層的 Oozie。這里我們不去細究各種 DAG 實現(xiàn)之間的區(qū)別,不過對于當時的 Tez 和 Oozie 來說,大多還是批處理的任務。

接下來就是以 Spark 為代表的第三代的計算引擎。第三代計算引擎的特點主要是 Job 內(nèi)部的 DAG 支持(不跨越 Job),以及強調(diào)的實時計算。在這里,很多人也會認為第三代計算引擎也能夠很好的運行批處理的 Job。

隨著第三代計算引擎的出現(xiàn),促進了上層應用快速發(fā)展,例如各種迭代計算的性能以及對流計算和 SQL 等的支持。Flink 的誕生就被歸在了第四代。這應該主要表現(xiàn)在 Flink 對流計算的支持,以及更一步的實時性上面。當然 Flink 也可以支持 Batch 的任務,以及 DAG 的運算。

總結:

第 1 代:Hadoop MapReduc 批處理 Mapper、Reducer 2;

第 2 代:DAG 框架(Oozie 、Tez),Tez + MapReduce 批處理 1 個 Tez = MR(1) + MR(2) + ．．． + MR(n) 相比 MR 效率有所提升;

第 3 代:Spark 批處理、流處理、SQL 高層 API 支持 自帶 DAG 內(nèi)存迭代計算、性能較之前大幅提;

第 4 代:Flink 批處理、流處理、SQL 高層 API 支持 自帶 DAG 流式計算性能更高、可靠性更高。

2． 什么是 Flink

Flink 起源于 Stratosphere 項目,Stratosphere 是在 2010～2014 年由 3 所地處柏林的大學和歐洲的一些其他的大學共同進行的研究項目,2014 年 4 月 Stratosphere 的代碼被復制并捐贈給了 Apache 軟件基金會,參加這個孵化項目的初始成員是 Stratosphere 系統(tǒng)的核心開發(fā)人員,2014 年 12 月,Flink 一躍成為 Apache 軟件基金會的頂級項目。

在德語中,Flink 一詞表示快速和靈巧,項目采用一只松鼠的彩色圖案作為 logo,這不僅是因為松鼠具有快速和靈巧的特點,還因為柏林的松鼠有一種迷人的紅棕色,而 Flink 的松鼠 logo 擁有可愛的尾巴,尾巴的顏色與 Apache 軟件基金會的 logo 顏色相呼應,也就是說,這是一只 Apache 風格的松鼠。

Flink 主頁在其頂部展示了該項目的理念:“Apache Flink 是為分布式、高性能、隨時可用以及準確的流處理應用程序打造的開源流處理框架”。

Apache Flink 是一個框架和分布式處理引擎,用于對無界和有界數(shù)據(jù)流進行有狀態(tài)計算。Flink 被設計在所有常見的集群環(huán)境中運行,以內(nèi)存執(zhí)行速度和任意規(guī)模來執(zhí)行計算。

3． Flink 流處理特性

支持高吞吐、低延遲、高性能的流處理

支持帶有事件時間的窗口(Window)操作

支持有狀態(tài)計算的 Exactly-once 語義

支持高度靈活的窗口(Window)操作,支持基于 time、count、session,以及 data-driven 的窗口操作

支持具有 Backpressure 功能的持續(xù)流模型

支持基于輕量級分布式快照(Snapshot)實現(xiàn)的容錯

一個運行時同時支持 Batch on Streaming 處理和 Streaming 處理

Flink 在 JVM 內(nèi)部實現(xiàn)了自己的內(nèi)存管理

支持迭代計算

支持程序自動優(yōu)化:避免特定情況下 Shuffle、排序等昂貴操作,中間結果有必要進行緩存

4． Flink 基石

Flink 之所以能這么流行,離不開它最重要的四個基石:Checkpoint、State、Time、Window。

首先是 Checkpoint 機制,這是 Flink 最重要的一個特性。Flink 基于Chandy-Lamport算法實現(xiàn)了一個分布式的一致性的快照,從而提供了一致性的語義。Chandy-Lamport 算法實際上在 1985 年的時候已經(jīng)被提出來,但并沒有被很廣泛的應用,而 Flink 則把這個算法發(fā)揚光大了。

Spark 最近在實現(xiàn) Continue streaming,Continue streaming 的目的是為了降低它處理的延時,其也需要提供這種一致性的語義,最終采用 Chandy-Lamport 這個算法,說明 Chandy-Lamport 算法在業(yè)界得到了一定的肯定。

提供了一致性的語義之后,Flink 為了讓用戶在編程時能夠更輕松、更容易地去管理狀態(tài),還提供了一套非常簡單明了的 State API,包括里面的有 ValueState、ListState、MapState,近期添加了 BroadcastState,使用 State API 能夠自動享受到這種一致性的語義。

除此之外,Flink 還實現(xiàn)了 Watermark 的機制,能夠支持基于事件的時間的處理,或者說基于系統(tǒng)時間的處理,能夠容忍數(shù)據(jù)的延時、容忍數(shù)據(jù)的遲到、容忍亂序的數(shù)據(jù)。

另外流計算中一般在對流數(shù)據(jù)進行操作之前都會先進行開窗,即基于一個什么樣的窗口上做這個計算。Flink 提供了開箱即用的各種窗口,比如滑動窗口、滾動窗口、會話窗口以及非常靈活的自定義的窗口。

5． 批處理與流處理

批處理的特點是有界、持久、大量,批處理非常適合需要訪問全套記錄才能完成的計算工作,一般用于離線統(tǒng)計。流處理的特點是無界、實時,流處理方式無需針對整個數(shù)據(jù)集執(zhí)行操作,而是對通過系統(tǒng)傳輸?shù)拿總�數(shù)據(jù)項執(zhí)行操作,一般用于實時統(tǒng)計。

在 Spark 生態(tài)體系中,對于批處理和流處理采用了不同的技術框架,批處理由 SparkSQL 實現(xiàn),流處理由 Spark Streaming 實現(xiàn),這也是大部分框架采用的策略,使用獨立的處理器實現(xiàn)批處理和流處理,而 Flink 可以同時實現(xiàn)批處理和流處理。

Flink 是如何同時實現(xiàn)批處理與流處理的呢?答案是,Flink 將批處理(即處理有限的靜態(tài)數(shù)據(jù))視作一種特殊的流處理。

Flink 的核心計算架構是下圖中的 Flink Runtime 執(zhí)行引擎,它是一個分布式系統(tǒng),能夠接受數(shù)據(jù)流程序并在一臺或多臺機器上以容錯方式執(zhí)行。

Flink Runtime 執(zhí)行引擎可以作為 YARN(Yet Another Resource Negotiator)的應用程序在集群上運行,也可以在 Mesos 集群上運行,還可以在單機上運行(這對于調(diào)試 Flink 應用程序來說非常有用)。

上圖為 Flink 技術棧的核心組成部分,值得一提的是,Flink 分別提供了面向流式處理的接口(DataStream API)和面向批處理的接口(DataSet API)。因此,Flink 既可以完成流處理,也可以完成批處理。Flink 支持的拓展庫涉及機器學習(FlinkML)、復雜事件處理(CEP)、以及圖計算(Gelly),還有分別針對流處理和批處理的 Table API。

能被 Flink Runtime 執(zhí)行引擎接受的程序很強大,但是這樣的程序有著冗長的代碼,編寫起來也很費力,基于這個原因,Flink 提供了封裝在 Runtime 執(zhí)行引擎之上的 API,以幫助用戶方便地生成流式計算程序。Flink 提供了用于流處理的 DataStream API 和用于批處理的 DataSet API。值得注意的是,盡管 Flink Runtime 執(zhí)行引擎是基于流處理的,但是 DataSet API 先于 DataStream API 被開發(fā)出來,這是因為工業(yè)界對無限流處理的需求在 Flink 誕生之初并不大。

DataStream API 可以流暢地分析無限數(shù)據(jù)流,并且可以用 Java 或者 Scala 等來實現(xiàn)。開發(fā)人員需要基于一個叫 DataStream 的數(shù)據(jù)結構來開發(fā),這個數(shù)據(jù)結構用于表示永不停止的分布式數(shù)據(jù)流。

Flink 的分布式特點體現(xiàn)在它能夠在成百上千臺機器上運行,它將大型的計算任務分成許多小的部分,每個機器執(zhí)行一部分。Flink 能夠自動地確保發(fā)生機器故障或者其他錯誤時計算能夠持續(xù)進行,或者在修復 bug 或進行版本升級后有計劃地再執(zhí)行一次。這種能力使得開發(fā)人員不需要擔心運行失敗。Flink 本質(zhì)上使用容錯性數(shù)據(jù)流,這使得開發(fā)人員可以分析持續(xù)生成且永遠不結束的數(shù)據(jù)(即流處理)。

二、Flink 部署及啟動

Flink 支持多種安裝模式:

local(本地)——單機模式,一般不使用;

standalone——獨立模式,Flink 自帶集群,開發(fā)測試環(huán)境使用;

yarn——計算資源統(tǒng)一由 Hadoop YARN 管理,生產(chǎn)環(huán)境使用。

Flink 集群的安裝不屬于本文檔的范疇,如安裝 Flink,可自行搜索資料進行安裝。

本節(jié)重點在 Flink 的 Yarn 部署模式。

在一個企業(yè)中,為了最大化的利用集群資源,一般都會在一個集群中同時運行多種類型的 Workload,可以使用 YARN 來管理所有計算資源。

1． Flink 在 Yarn 上的部署架構

從圖中可以看出,Yarn 的客戶端需要獲取 hadoop 的配置信息,連接 Yarn 的 ResourceManager。所以要設置 YARN_CONF_DIR 或者 HADOOP_CONF_DIR 或者 HADOOP_CONF_PATH,只要設置了其中一個環(huán)境變量,就會被讀取。如果讀取上述的變量失敗了,那么將會選擇 hadoop_home 的環(huán)境變量,會嘗試加載$HADOOP_HOME/etc/hadoop 的配置文件。

當啟動一個 Flink Yarn 會話時,客戶端首先會檢查本次請求的資源(存儲、計算)是否足夠。資源足夠將會上傳包含 HDFS 及 Flink 的配置信息和 Flink 的 jar 包到 HDFS;

客戶端向 RM 發(fā)起請求;

RM 向 NM 發(fā)請求指令,創(chuàng)建 container,并從 HDFS 中下載 jar 以及配置文件;

啟動 ApplicationMaster 和 jobmanager,將 jobmanager 的地址信息寫到配置文件中,再發(fā)到 hdfs 上;

同時,AM 向 RM 發(fā)送心跳注冊自己,申請資源(cpu、內(nèi)存);

創(chuàng)建 TaskManager 容器,從 HDFS 中下載 jar 包及配置文件并啟動;

各 task 任務通過 jobmanager 匯報自己的狀態(tài)和進度,AM 和 jobmanager 在一個容器上,AM 就能掌握各任務的運行狀態(tài),從而可以在任務失敗時,重新啟動任務;

任務完成后,AM 向 RM 注銷并關閉自己;

2． 啟動集群修改 hadoop 的配置參數(shù):vim etc/hadoop/yarn-site．xml

添加:

修改 Hadoop 的 yarn-site．xml,添加該配置表示內(nèi)存超過分配值,是否將任務殺掉。

默認為 true。運行 Flink 程序,很容易內(nèi)存超標,這個時候 yarn 會自動殺掉 job。

修改全局變量 /etc/profile:

添加:export HADOOP_CONF_DIR=/export/servers/hadoop/etc/Hadoop

YARN_CONF_DIR 或者 HADOOP_CONF_DIR 必須將環(huán)境變量設置為讀取 YARN 和 HDFS 配置

啟動 HDFS、zookeeper(如果是外置 zookeeper)、YARN 集群;

使用 yarn-session 的模式提交作業(yè)。

Yarn Session 模式提交作業(yè)有兩種方式:yarn-session 和 yarn-cluster

3． 模式一: yarn-session

特點:

使用 Flink 中的 yarn-session(yarn 客戶端),會啟動兩個必要服務 JobManager 和 TaskManagers;

客戶端通過 yarn-session 提交作業(yè);

yarn-session 會一直啟動,不停地接收客戶端提交的任務;

如果擁有有大量的小作業(yè),適合使用這種方式。

在 flink 目錄啟動 yarn-session:

bin/yarn-session．sh -n 2 -tm 800 -jm 800 -s 1 -d

-n 表示申請 2 個容器
-s 表示每個容器啟動多少個 slot 離模式,表示以后臺程
-tm 表示每個 TaskManager 申請 800M 內(nèi)存
-d 分序方式運行

使用 flink 提交任務:

bin/flink run examples/batch/WordCount．jar

如果程序運行完了,可以使用 yarn application -kill application_id 殺掉任務:

yarn application -kill application_1554377097889_0002

bin/yarn-session．sh -n 2 -tm 800 -s 1 -d 意思是:

同時向 Yarn 申請 3 個 container(即便只申請了兩個,因為 ApplicationMaster 和 Job Manager 有一個額外的容器。一旦將 Flink 部署到 YARN 群集中,它就會顯示 Job Manager 的連接詳細信息),其中 2 個 Container 啟動 TaskManager(-n 2),每個 TaskManager 擁有兩個 Task Slot(-s 1),并且向每個 TaskManager 的 Container 申請 800M 的內(nèi)存,以及一個 ApplicationMaster(Job Manager)。

4． 模式二: yarn-cluster

特點:

直接提交任務給 YARN;

大作業(yè),適合使用這種方式;

會自動關閉 session。

使用 flink 直接提交任務:

bin/flink run -m yarn-cluster -yn 2 -yjm 800 -ytm 800 /export/servers/flink-1．6．0/examples/batch/WordCount．jar

-yn 表示 TaskManager 的個數(shù)

注意:

在創(chuàng)建集群的時候,集群的配置參數(shù)就寫好了,但是往往因為業(yè)務需要,要更改一些配置參數(shù),這個時候可以不必因為一個實例的提交而修改 conf/flink-conf．yaml;

可以通過:-D

-Dfs．overwrite-files=true -Dtaskmanager．network．numberOfBuffers=16368

如果使用的是 flink on yarn 方式,想切換回 standalone 模式的話,需要刪除:/tmp/．yarn-properties-root,因為默認查找當前 yarn 集群中已有的 yarn-session 信息中的 jobmanager。三、Flink 運行架構1． Flink 程序結構

Flink 程序的基本構建塊是流和轉換(請注意,Flink 的 DataSet API 中使用的 DataSet 也是內(nèi)部流 )。從概念上講,流是(可能永無止境的)數(shù)據(jù)記錄流,而轉換是將一個或多個流作為一個或多個流的操作。輸入,并產(chǎn)生一個或多個輸出流。

Flink 應用程序結構就是如上圖所示:

Source: 數(shù)據(jù)源,Flink 在流處理和批處理上的 source 大概有 4 類:基于本地集合的 source、基于文件的 source、基于網(wǎng)絡套接字的 source、自定義的 source。自定義的 source 常見的有 Apache kafka、RabbitMQ 等,當然你也可以定義自己的 source。

Transformation:數(shù)據(jù)轉換的各種操作,有 Map / FlatMap / Filter / KeyBy / Reduce / Fold / Aggregations / Window / WindowAll / Union / Window join / Split / Select等,操作很多,可以將數(shù)據(jù)轉換計算成你想要的數(shù)據(jù)。

Sink:接收器,Flink 將轉換計算后的數(shù)據(jù)發(fā)送的地點 ,你可能需要存儲下來,Flink 常見的 Sink 大概有如下幾類:寫入文件、打印出來、寫入 socket 、自定義的 sink 。自定義的 sink 常見的有 Apache kafka、RabbitMQ、MySQL、ElasticSearch、Apache Cassandra、Hadoop FileSystem 等,同理你也可以定義自己的 sink。

2． Flink 并行數(shù)據(jù)流

Flink 程序在執(zhí)行的時候,會被映射成一個 Streaming Dataflow,一個 Streaming Dataflow 是由一組 Stream 和 Transformation Operator 組成的。在啟動時從一個或多個 Source Operator 開始,結束于一個或多個 Sink Operator。

Flink 程序本質(zhì)上是并行的和分布式的,在執(zhí)行過程中,一個流(stream)包含一個或多個流分區(qū),而每一個 operator 包含一個或多個 operator 子任務。操作子任務間彼此獨立,在不同的線程中執(zhí)行,甚至是在不同的機器或不同的容器上。operator 子任務的數(shù)量是這一特定 operator 的并行度。相同程序中的不同 operator 有不同級別的并行度。

一個 Stream 可以被分成多個 Stream 的分區(qū),也就是 Stream Partition。一個 Operator 也可以被分為多個 Operator Subtask。如上圖中,Source 被分成 Source1 和 Source2,它們分別為 Source 的 Operator Subtask。每一個 Operator Subtask 都是在不同的線程當中獨立執(zhí)行的。一個 Operator 的并行度,就等于 Operator Subtask 的個數(shù)。上圖 Source 的并行度為 2。而一個 Stream 的并行度就等于它生成的 Operator 的并行度。

數(shù)據(jù)在兩個 operator 之間傳遞的時候有兩種模式:

One to One 模式:兩個 operator 用此模式傳遞的時候,會保持數(shù)據(jù)的分區(qū)數(shù)和數(shù)據(jù)的排序;如上圖中的 Source1 到 Map1,它就保留的 Source 的分區(qū)特性,以及分區(qū)元素處理的有序性。

Redistributing (重新分配)模式:這種模式會改變數(shù)據(jù)的分區(qū)數(shù);每個一個 operator subtask 會根據(jù)選擇 transformation 把數(shù)據(jù)發(fā)送到不同的目標 subtasks,比如 keyBy()會通過 hashcode 重新分區(qū),broadcast()和 rebalance()方法會隨機重新分區(qū);

3． Task 和 Operator chain

Flink的所有操作都稱之為Operator,客戶端在提交任務的時候會對Operator進行優(yōu)化操作,能進行合并的Operator會被合并為一個Operator,合并后的Operator稱為Operator chain,實際上就是一個執(zhí)行鏈,每個執(zhí)行鏈會在TaskManager上一個獨立的線程中執(zhí)行。

4． 任務調(diào)度與執(zhí)行

當Flink執(zhí)行executor會自動根據(jù)程序代碼生成DAG數(shù)據(jù)流圖;

ActorSystem創(chuàng)建Actor將數(shù)據(jù)流圖發(fā)送給JobManager中的Actor;

JobManager會不斷接收TaskManager的心跳消息,從而可以獲取到有效的TaskManager;

JobManager通過調(diào)度器在TaskManager中調(diào)度執(zhí)行Task(在Flink中,最小的調(diào)度單元就是task,對應就是一個線程);

在程序運行過程中,task與task之間是可以進行數(shù)據(jù)傳輸?shù)摹?/p>

Job Client:

主要職責是提交任務, 提交后可以結束進程, 也可以等待結果返回;Job Client 不是 Flink 程序執(zhí)行的內(nèi)部部分,但它是任務執(zhí)行的起點;Job Client 負責接受用戶的程序代碼,然后創(chuàng)建數(shù)據(jù)流,將數(shù)據(jù)流提交給 Job Manager 以便進一步執(zhí)行。執(zhí)行完成后,Job Client 將結果返回給用戶。

JobManager:

主要職責是調(diào)度工作并協(xié)調(diào)任務做檢查點;集群中至少要有一個 master,master 負責調(diào)度 task,協(xié)調(diào)checkpoints 和容錯;高可用設置的話可以有多個 master,但要保證一個是 leader, 其他是standby;Job Manager 包含 Actor System、Scheduler、CheckPoint三個重要的組件;JobManager從客戶端接收到任務以后, 首先生成優(yōu)化過的執(zhí)行計劃, 再調(diào)度到TaskManager中執(zhí)行。

TaskManager:

主要職責是從JobManager處接收任務, 并部署和啟動任務, 接收上游的數(shù)據(jù)并處理;Task Manager 是在 JVM 中的一個或多個線程中執(zhí)行任務的工作節(jié)點;TaskManager在創(chuàng)建之初就設置好了Slot, 每個Slot可以執(zhí)行一個任務。5． 任務槽和槽共享

每個TaskManager是一個JVM的進程, 可以在不同的線程中執(zhí)行一個或多個子任務。為了控制一個worker能接收多少個task。worker通過task slot來進行控制(一個worker至少有一個task slot)。

1) 任務槽

每個task slot表示TaskManager擁有資源的一個固定大小的子集。

flink將進程的內(nèi)存進行了劃分到多個slot中。

圖中有2個TaskManager,每個TaskManager有3個slot的,每個slot占有1/3的內(nèi)存。

內(nèi)存被劃分到不同的slot之后可以獲得如下好處:

TaskManager最多能同時并發(fā)執(zhí)行的任務是可以控制的,那就是3個,因為不能超過slot的數(shù)量。

slot有獨占的內(nèi)存空間,這樣在一個TaskManager中可以運行多個不同的作業(yè),作業(yè)之間不受影響。

2) 槽共享

默認情況下,Flink允許子任務共享插槽,即使它們是不同任務的子任務,只要它們來自同一個作業(yè)。結果是一個槽可以保存作業(yè)的整個管道。允許插槽共享有兩個主要好處:

只需計算Job中最高并行度(parallelism)的task slot,只要這個滿足,其他的job也都能滿足。

資源分配更加公平,如果有比較空閑的slot可以將更多的任務分配給它。圖中若沒有任務槽共享,負載不高的Source/Map等subtask將會占據(jù)許多資源,而負載較高的窗口subtask則會缺乏資源。

有了任務槽共享,可以將基本并行度(base parallelism)從2提升到6．提高了分槽資源的利用率。同時它還可以保障TaskManager給subtask的分配的slot方案更加公平。