作者：陳守元、戴資力

成都創(chuàng)新互聯(lián)公司長期為上千多家客戶提供的網(wǎng)站建設(shè)服務(wù)，團(tuán)隊(duì)從業(yè)經(jīng)驗(yàn)10年，關(guān)注不同地域、不同群體，并針對(duì)不同對(duì)象提供差異化的產(chǎn)品和服務(wù)；打造開放共贏平臺(tái)，與合作伙伴共同營造健康的互聯(lián)網(wǎng)生態(tài)環(huán)境。為米易企業(yè)提供專業(yè)的成都網(wǎng)站制作、做網(wǎng)站，米易網(wǎng)站改版等技術(shù)服務(wù)。擁有十年豐富建站經(jīng)驗(yàn)和眾多成功案例,為您定制開發(fā)。

一、Apache Flink 的定義、架構(gòu)及原理

Apache Flink 是一個(gè)分布式大數(shù)據(jù)處理引擎，可對(duì)有限數(shù)據(jù)流和無限數(shù)據(jù)流進(jìn)行有狀態(tài)或無狀態(tài)的計(jì)算，能夠部署在各種集群環(huán)境，對(duì)各種規(guī)模大小的數(shù)據(jù)進(jìn)行快速計(jì)算。

1. Flink Application

了解 Flink 應(yīng)用開發(fā)需要先理解 Flink 的 Streams、State、Time 等基礎(chǔ)處理語義以及 Flink 兼顧靈活性和方便性的多層次 API。

• Streams：流，分為有限數(shù)據(jù)流與無限數(shù)據(jù)流，unbounded stream 是有始無終的數(shù)據(jù)流，即無限數(shù)據(jù)流；而 bounded stream 是限定大小的有始有終的數(shù)據(jù)集合，即有限數(shù)據(jù)流，二者的區(qū)別在于無限數(shù)據(jù)流的數(shù)據(jù)會(huì)隨時(shí)間的推演而持續(xù)增加，計(jì)算持續(xù)進(jìn)行且不存在結(jié)束的狀態(tài)，相對(duì)的有限數(shù)據(jù)流數(shù)據(jù)大小固定，計(jì)算最終會(huì)完成并處于結(jié)束的狀態(tài)。

• State，狀態(tài)是計(jì)算過程中的數(shù)據(jù)信息，在容錯(cuò)恢復(fù)和 Checkpoint 中有重要的作用，流計(jì)算在本質(zhì)上是 Incremental Processing，因此需要不斷查詢保持狀態(tài)；另外，為了確保 Exactly- once 語義，需要數(shù)據(jù)能夠?qū)懭氲綘顟B(tài)中；而持久化存儲(chǔ)，能夠保證在整個(gè)分布式系統(tǒng)運(yùn)行失敗或者掛掉的情況下做到 Exactly- once，這是狀態(tài)的另外一個(gè)價(jià)值。

• Time，分為 Event time、Ingestion time、Processing time，F(xiàn)link 的無限數(shù)據(jù)流是一個(gè)持續(xù)的過程，時(shí)間是我們判斷業(yè)務(wù)狀態(tài)是否滯后，數(shù)據(jù)處理是否及時(shí)的重要依據(jù)。

• API，API 通常分為三層，由上而下可分為 SQL / Table API、DataStream API、ProcessFunction 三層，API 的表達(dá)能力及業(yè)務(wù)抽象能力都非常強(qiáng)大，但越接近 SQL 層，表達(dá)能力會(huì)逐步減弱，抽象能力會(huì)增強(qiáng)，反之，ProcessFunction 層 API 的表達(dá)能力非常強(qiáng)，可以進(jìn)行多種靈活方便的操作，但抽象能力也相對(duì)越小。

2.Flink Architecture

在架構(gòu)部分，主要分為以下四點(diǎn)：

第一， Flink 具備統(tǒng)一的框架處理有界和×××兩種數(shù)據(jù)流的能力

第二， 部署靈活，F(xiàn)link 底層支持多種資源調(diào)度器，包括 Yarn、Kubernetes 等。Flink 自身帶的 Standalone 的調(diào)度器，在部署上也十分靈活。

第三， 極高的可伸縮性，可伸縮性對(duì)于分布式系統(tǒng)十分重要，阿里巴巴雙11大屏采用 Flink 處理海量數(shù)據(jù)，使用過程中測得 Flink 峰值可達(dá) 17 億/秒。

第四， 極致的流式處理性能。Flink 相對(duì)于 Storm 大的特點(diǎn)是將狀態(tài)語義完全抽象到框架中，支持本地狀態(tài)讀取，避免了大量網(wǎng)絡(luò) IO，可以極大提升狀態(tài)存取的性能。

3.Flink Operation

后面會(huì)有專門課程講解，此處簡單分享 Flink 關(guān)于運(yùn)維及業(yè)務(wù)監(jiān)控的內(nèi)容：

• Flink 具備 7 X 24 小時(shí)高可用的 SOA（面向服務(wù)架構(gòu)），原因是在實(shí)現(xiàn)上 Flink 提供了一致性的 Checkpoint。Checkpoint 是 Flink 實(shí)現(xiàn)容錯(cuò)機(jī)制的核心，它周期性的記錄計(jì)算過程中 Operator 的狀態(tài)，并生成快照持久化存儲(chǔ)。當(dāng) Flink 作業(yè)發(fā)生故障崩潰時(shí)，可以有選擇的從 Checkpoint 中恢復(fù)，保證了計(jì)算的一致性。

• Flink 本身提供監(jiān)控、運(yùn)維等功能或接口，并有內(nèi)置的 WebUI，對(duì)運(yùn)行的作業(yè)提供 DAG 圖以及各種 Metric 等，協(xié)助用戶管理作業(yè)狀態(tài)。

4.Flink 的應(yīng)用場景

4.1 Flink 的應(yīng)用場景：Data Pipeline

Data Pipeline 的核心場景類似于數(shù)據(jù)搬運(yùn)并在搬運(yùn)的過程中進(jìn)行部分?jǐn)?shù)據(jù)清洗或者處理，而整個(gè)業(yè)務(wù)架構(gòu)圖的左邊是 Periodic ETL，它提供了流式 ETL 或者實(shí)時(shí) ETL，能夠訂閱消息隊(duì)列的消息并進(jìn)行處理，清洗完成后實(shí)時(shí)寫入到下游的 Database 或 File system 中。場景舉例：

• 實(shí)時(shí)數(shù)倉

當(dāng)下游要構(gòu)建實(shí)時(shí)數(shù)倉時(shí)，上游則可能需要實(shí)時(shí)的 Stream ETL。這個(gè)過程會(huì)進(jìn)行實(shí)時(shí)清洗或擴(kuò)展數(shù)據(jù)，清洗完成后寫入到下游的實(shí)時(shí)數(shù)倉的整個(gè)鏈路中，可保證數(shù)據(jù)查詢的時(shí)效性，形成實(shí)時(shí)數(shù)據(jù)采集、實(shí)時(shí)數(shù)據(jù)處理以及下游的實(shí)時(shí) Query。

• 搜索引擎推薦

搜索引擎這塊以淘寶為例，當(dāng)賣家上線新商品時(shí)，后臺(tái)會(huì)實(shí)時(shí)產(chǎn)生消息流，該消息流經(jīng)過 Flink 系統(tǒng)時(shí)會(huì)進(jìn)行數(shù)據(jù)的處理、擴(kuò)展。然后將處理及擴(kuò)展后的數(shù)據(jù)生成實(shí)時(shí)索引，寫入到搜索引擎中。這樣當(dāng)淘寶賣家上線新商品時(shí)，能在秒級(jí)或者分鐘級(jí)實(shí)現(xiàn)搜索引擎的搜索。

4.2 Flink 應(yīng)用場景：Data Analytics

Data Analytics，如圖，左邊是 Batch Analytics，右邊是 Streaming Analytics。Batch Analysis 就是傳統(tǒng)意義上使用類似于 Map Reduce、Hive、Spark Batch 等，對(duì)作業(yè)進(jìn)行分析、處理、生成離線報(bào)表，Streaming Analytics 使用流式分析引擎如 Storm，F(xiàn)link 實(shí)時(shí)處理分析數(shù)據(jù)，應(yīng)用較多的場景如實(shí)時(shí)大屏、實(shí)時(shí)報(bào)表。

4.3 Flink 應(yīng)用場景：Data Driven

從某種程度上來說，所有的實(shí)時(shí)的數(shù)據(jù)處理或者是流式數(shù)據(jù)處理都是屬于 Data Driven，流計(jì)算本質(zhì)上是 Data Driven 計(jì)算。應(yīng)用較多的如風(fēng)控系統(tǒng)，當(dāng)風(fēng)控系統(tǒng)需要處理各種各樣復(fù)雜的規(guī)則時(shí)，Data Driven 就會(huì)把處理的規(guī)則和邏輯寫入到Datastream 的 API 或者是 ProcessFunction 的 API 中，然后將邏輯抽象到整個(gè) Flink 引擎中，當(dāng)外面的數(shù)據(jù)流或者是事件進(jìn)入就會(huì)觸發(fā)相應(yīng)的規(guī)則，這就是 Data Driven 的原理。在觸發(fā)某些規(guī)則后，Data Driven 會(huì)進(jìn)行處理或者是進(jìn)行預(yù)警，這些預(yù)警會(huì)發(fā)到下游產(chǎn)生業(yè)務(wù)通知，這是 Data Driven 的應(yīng)用場景，Data Driven 在應(yīng)用上更多應(yīng)用于復(fù)雜事件的處理。

二、「有狀態(tài)的流式處理」概念解析

1.傳統(tǒng)批處理

傳統(tǒng)批處理方法是持續(xù)收取數(shù)據(jù)，以時(shí)間作為劃分多個(gè)批次的依據(jù)，再周期性地執(zhí)行批次運(yùn)算。但假設(shè)需要計(jì)算每小時(shí)出現(xiàn)事件轉(zhuǎn)換的次數(shù)，如果事件轉(zhuǎn)換跨越了所定義的時(shí)間劃分，傳統(tǒng)批處理會(huì)將中介運(yùn)算結(jié)果帶到下一個(gè)批次進(jìn)行計(jì)算；除此之外，當(dāng)出現(xiàn)接收到的事件順序顛倒情況下，傳統(tǒng)批處理仍會(huì)將中介狀態(tài)帶到下一批次的運(yùn)算結(jié)果中，這種處理方式也不盡如人意。

2.理想方法

第一點(diǎn)，要有理想方法，這個(gè)理想方法是引擎必須要有能力可以累積狀態(tài)和維護(hù)狀態(tài)，累積狀態(tài)代表著過去歷史中接收過的所有事件，會(huì)影響到輸出。

第二點(diǎn)，時(shí)間，時(shí)間意味著引擎對(duì)于數(shù)據(jù)完整性有機(jī)制可以操控，當(dāng)所有數(shù)據(jù)都完全接受到后，輸出計(jì)算結(jié)果。

第三點(diǎn)，理想方法模型需要實(shí)時(shí)產(chǎn)生結(jié)果，但更重要的是采用新的持續(xù)性數(shù)據(jù)處理模型來處理實(shí)時(shí)數(shù)據(jù)，這樣才最符合 continuous data 的特性。

3.流式處理

流式處理簡單來講即有一個(gè)無窮無盡的數(shù)據(jù)源在持續(xù)收取數(shù)據(jù)，以代碼作為數(shù)據(jù)處理的基礎(chǔ)邏輯，數(shù)據(jù)源的數(shù)據(jù)經(jīng)過代碼處理后產(chǎn)生出結(jié)果，然后輸出，這就是流式處理的基本原理。

4.分布式流式處理

假設(shè) Input Streams 有很多個(gè)使用者，每個(gè)使用者都有自己的 ID，如果計(jì)算每個(gè)使用者出現(xiàn)的次數(shù)，我們需要讓同一個(gè)使用者的出現(xiàn)事件流到同一運(yùn)算代碼，這跟其他批次需要做 group by 是同樣的概念，所以跟 Stream 一樣需要做分區(qū)，設(shè)定相應(yīng)的 key，然后讓同樣的 key 流到同一個(gè) computation instance 做同樣的運(yùn)算。

5.有狀態(tài)分布式流式處理

如圖，上述代碼中定義了變數(shù) X，X 在數(shù)據(jù)處理過程中會(huì)進(jìn)行讀和寫，在最后輸出結(jié)果時(shí)，可以依據(jù)變數(shù) X 決定輸出的內(nèi)容，即狀態(tài) X 會(huì)影響最終的輸出結(jié)果。這個(gè)過程中，第一個(gè)重點(diǎn)是先進(jìn)行了狀態(tài) co-partitioned key by，同樣的 key 都會(huì)流到 computation instance，與使用者出現(xiàn)次數(shù)的原理相同，次數(shù)即所謂的狀態(tài)，這個(gè)狀態(tài)一定會(huì)跟同一個(gè) key 的事件累積在同一個(gè) computation instance。

相當(dāng)于根據(jù)輸入流的 key 重新分區(qū)的 狀態(tài)，當(dāng)分區(qū)進(jìn)入 stream 之后，這個(gè) stream 會(huì)累積起來的狀態(tài)也變成 copartiton 了。第二個(gè)重點(diǎn)是 embeded local state backend。有狀態(tài)分散式流式處理的引擎，狀態(tài)可能會(huì)累積到非常大，當(dāng) key 非常多時(shí)，狀態(tài)可能就會(huì)超出單一節(jié)點(diǎn)的 memory 的負(fù)荷量，這時(shí)候狀態(tài)必須有狀態(tài)后端去維護(hù)它；在這個(gè)狀態(tài)后端在正常狀況下，用 in-memory 維護(hù)即可。

三、Apache Flink 的優(yōu)勢

1.狀態(tài)容錯(cuò)

當(dāng)我們考慮狀態(tài)容錯(cuò)時(shí)難免會(huì)想到精確一次的狀態(tài)容錯(cuò)，應(yīng)用在運(yùn)算時(shí)累積的狀態(tài)，每筆輸入的事件反映到狀態(tài)，更改狀態(tài)都是精確一次，如果修改超過一次的話也意味著數(shù)據(jù)引擎產(chǎn)生的結(jié)果是不可靠的。

• 如何確保狀態(tài)擁有精確一次（Exactly-once guarantee）的容錯(cuò)保證？

• 如何在分散式場景下替多個(gè)擁有本地狀態(tài)的運(yùn)算子產(chǎn)生一個(gè)全域一致的快照（Global consistent snapshot）？

• 更重要的是，如何在不中斷運(yùn)算的前提下產(chǎn)生快照？

1.1 簡單場景的精確一次容錯(cuò)方法

還是以使用者出現(xiàn)次數(shù)來看，如果某個(gè)使用者出現(xiàn)的次數(shù)計(jì)算不準(zhǔn)確，不是精確一次，那么產(chǎn)生的結(jié)果是無法作為參考的。在考慮精確的容錯(cuò)保證前，我們先考慮最簡單的使用場景，如無限流的數(shù)據(jù)進(jìn)入，后面單一的 Process 進(jìn)行運(yùn)算，每處理完一筆計(jì)算即會(huì)累積一次狀態(tài)，這種情況下如果要確保 Process 產(chǎn)生精確一次的狀態(tài)容錯(cuò)，每處理完一筆數(shù)據(jù)，更改完?duì)顟B(tài)后進(jìn)行一次快照，快照包含在隊(duì)列中并與相應(yīng)的狀態(tài)進(jìn)行對(duì)比，完成一致的快照，就能確保精確一次。

1.2 分布式狀態(tài)容錯(cuò)

Flink 作為分布式的處理引擎，在分布式的場景下，進(jìn)行多個(gè)本地狀態(tài)的運(yùn)算，只產(chǎn)生一個(gè)全域一致的快照，如需要在不中斷運(yùn)算值的前提下產(chǎn)生全域一致的快照，就涉及到分散式狀態(tài)容錯(cuò)。

• Global consistent snapshot

關(guān)于 Global consistent snapshot，當(dāng) Operator 在分布式的環(huán)境中，在各個(gè)節(jié)點(diǎn)做運(yùn)算，首先產(chǎn)生 Global consistent snapshot 的方式就是處理每一筆數(shù)據(jù)的快照點(diǎn)是連續(xù)的，這筆運(yùn)算流過所有的運(yùn)算值，更改完所有的運(yùn)算值后，能夠看到每一個(gè)運(yùn)算值的狀態(tài)與該筆運(yùn)算的位置，即可稱為 consistent snapshot，當(dāng)然，Global consistent snapshot 也是簡易場景的延伸。

• 容錯(cuò)恢復(fù)

首先了解一下 Checkpoint，上面提到連續(xù)性快照每個(gè) Operator 運(yùn)算值本地的狀態(tài)后端都要維護(hù)狀態(tài)，也就是每次將產(chǎn)生檢查點(diǎn)時(shí)會(huì)將它們傳入共享的 DFS 中。當(dāng)任何一個(gè) Process 掛掉后，可以直接從三個(gè)完整的 Checkpoint 將所有的運(yùn)算值的狀態(tài)恢復(fù)，重新設(shè)定到相應(yīng)位置。Checkpoint 的存在使整個(gè) Process 能夠?qū)崿F(xiàn)分散式環(huán)境中的 Exactly-once。

1.3 分散式快照（Distributed Snapshots）方法

關(guān)于 Flink 如何在不中斷運(yùn)算的狀況下持續(xù)產(chǎn)生 Global consistent snapshot，其方式是基于用 simple lamport 演算法機(jī)制下延伸的。已知的一個(gè)點(diǎn) Checkpoint barrier， Flink 在某個(gè) Datastream 中會(huì)一直安插 Checkpoint barrier，Checkpoint barrier 也會(huì) N — 1等等，Checkpoint barrier N 代表著所有在這個(gè)范圍里面的數(shù)據(jù)都是Checkpoint barrier N。

舉例：假設(shè)現(xiàn)在需要產(chǎn)生 Checkpoint barrier N，但實(shí)際上在 Flink 中是由 job manager 觸發(fā) Checkpoint，Checkpoint 被觸發(fā)后開始從數(shù)據(jù)源產(chǎn)生 Checkpoint barrier。當(dāng) job 開始做 Checkpoint barrier N 的時(shí)候，可以理解為 Checkpoint barrier N 需要逐步填充左下角的表格。

如圖，當(dāng)部分事件標(biāo)為紅色，Checkpoint barrier N 也是紅色時(shí)，代表著這些數(shù)據(jù)或事件都由 Checkpoint barrier N 負(fù)責(zé)。Checkpoint barrier N 后面白色部分的數(shù)據(jù)或事件則不屬于 Checkpoint barrier N。

在以上的基礎(chǔ)上，當(dāng)數(shù)據(jù)源收到 Checkpoint barrier N 之后會(huì)先將自己的狀態(tài)保存，以讀取 Kafka 資料為例，數(shù)據(jù)源的狀態(tài)就是目前它在 Kafka 分區(qū)的位置，這個(gè)狀態(tài)也會(huì)寫入到上面提到的表格中。下游的 Operator 1 會(huì)開始運(yùn)算屬于 Checkpoint barrier N 的數(shù)據(jù)，當(dāng) Checkpoint barrier N 跟著這些數(shù)據(jù)流動(dòng)到 Operator 1 之后,Operator 1 也將屬于 Checkpoint barrier N 的所有數(shù)據(jù)都反映在狀態(tài)中，當(dāng)收到 Checkpoint barrier N 時(shí)也會(huì)直接對(duì) Checkpoint 去做快照。

當(dāng)快照完成后繼續(xù)往下游走，Operator 2 也會(huì)接收到所有數(shù)據(jù)，然后搜索 Checkpoint barrier N 的數(shù)據(jù)并直接反映到狀態(tài)，當(dāng)狀態(tài)收到 Checkpoint barrier N 之后也會(huì)直接寫入到 Checkpoint N 中。以上過程到此可以看到 Checkpoint barrier N 已經(jīng)完成了一個(gè)完整的表格，這個(gè)表格叫做 Distributed Snapshots，即分布式快照。分布式快照可以用來做狀態(tài)容錯(cuò)，任何一個(gè)節(jié)點(diǎn)掛掉的時(shí)候可以在之前的 Checkpoint 中將其恢復(fù)。繼續(xù)以上 Process，當(dāng)多個(gè) Checkpoint 同時(shí)進(jìn)行，Checkpoint barrier N 已經(jīng)流到 job manager 2，F(xiàn)link job manager 可以觸發(fā)其他的 Checkpoint，比如 Checkpoint N + 1，Checkpoint N + 2 等等也同步進(jìn)行，利用這種機(jī)制，可以在不阻擋運(yùn)算的狀況下持續(xù)地產(chǎn)生 Checkpoint。

2.狀態(tài)維護(hù)

狀態(tài)維護(hù)即用一段代碼在本地維護(hù)狀態(tài)值，當(dāng)狀態(tài)值非常大時(shí)需要本地的狀態(tài)后端來支持。

如圖，在 Flink 程序中，可以采用 getRuntimeContext().getState(desc); 這組 API 去注冊狀態(tài)。Flink 有多種狀態(tài)后端，采用 API 注冊狀態(tài)后，讀取狀態(tài)時(shí)都是通過狀態(tài)后端來讀取的。Flink 有兩種不同的狀態(tài)值，也有兩種不同的狀態(tài)后端：

• JVM Heap狀態(tài)后端，適合數(shù)量較小的狀態(tài)，當(dāng)狀態(tài)量不大時(shí)就可以采用 JVM Heap 的狀態(tài)后端。JVM Heap 狀態(tài)后端會(huì)在每一次運(yùn)算值需要讀取狀態(tài)時(shí)，用 Java object read / writes 進(jìn)行讀或?qū)懀粫?huì)產(chǎn)生較大代價(jià)，但當(dāng) Checkpoint 需要將每一個(gè)運(yùn)算值的本地狀態(tài)放入 Distributed Snapshots 的時(shí)候，就需要進(jìn)行序列化了。

• RocksDB 狀態(tài)后端，它是一種 out of core 的狀態(tài)后端。在 Runtime 的本地狀態(tài)后端讓使用者去讀取狀態(tài)的時(shí)候會(huì)經(jīng)過磁盤，相當(dāng)于將狀態(tài)維護(hù)在磁盤里，與之對(duì)應(yīng)的代價(jià)可能就是每次讀取狀態(tài)時(shí)，都需要經(jīng)過序列化和反序列化的過程。當(dāng)需要進(jìn)行快照時(shí)只將應(yīng)用序列化即可，序列化后的數(shù)據(jù)直接傳輸?shù)街醒氲墓蚕?DFS 中。

Flink 目前支持以上兩種狀態(tài)后端，一種是純 memory 的狀態(tài)后端，另一種是有資源磁盤的狀態(tài)后端，在維護(hù)狀態(tài)時(shí)可以根據(jù)狀態(tài)的數(shù)量選擇相應(yīng)的狀態(tài)后端。

3.Event - Time

3.1 不同時(shí)間種類

在 Flink 及其他進(jìn)階的流式處理引擎出現(xiàn)之前，大數(shù)據(jù)處理引擎一直只支持 Processing-time 的處理。假設(shè)定義一個(gè)運(yùn)算 windows 的窗口，windows 運(yùn)算設(shè)定每小時(shí)進(jìn)行結(jié)算。以 Processing-time 進(jìn)行運(yùn)算時(shí)可以發(fā)現(xiàn)數(shù)據(jù)引擎將 3 點(diǎn)至 4 點(diǎn)間收到的數(shù)據(jù)做結(jié)算。實(shí)際上在做報(bào)表或者分析結(jié)果時(shí)是想了解真實(shí)世界中 3 點(diǎn)至 4 點(diǎn)之間實(shí)際產(chǎn)生數(shù)據(jù)的輸出結(jié)果，了解實(shí)際數(shù)據(jù)的輸出結(jié)果就必須采用 Event – Time 了。

如圖，Event - Time 相當(dāng)于事件，它在數(shù)據(jù)最源頭產(chǎn)生時(shí)帶有時(shí)間戳，后面都需要用時(shí)間戳來進(jìn)行運(yùn)算。用圖來表示，最開始的隊(duì)列收到數(shù)據(jù)，每小時(shí)對(duì)數(shù)據(jù)劃分一個(gè)批次，這就是 Event - Time Process 在做的事情。

3.2 Event - Time 處理

Event - Time 是用事件真實(shí)產(chǎn)生的時(shí)間戳去做 Re-bucketing，把對(duì)應(yīng)時(shí)間 3 點(diǎn)到 4 點(diǎn)的數(shù)據(jù)放在 3 點(diǎn)到 4 點(diǎn)的 Bucket，然后 Bucket 產(chǎn)生結(jié)果。所以 Event - Time 跟 Processing - time 的概念是這樣對(duì)比的存在。

Event - Time 的重要性在于記錄引擎輸出運(yùn)算結(jié)果的時(shí)間。簡單來說，流式引擎連續(xù) 24 小時(shí)在運(yùn)行、搜集資料，假設(shè) Pipeline 里有一個(gè) windows Operator 正在做運(yùn)算，每小時(shí)能產(chǎn)生結(jié)果，何時(shí)輸出 windows 的運(yùn)算值，這個(gè)時(shí)間點(diǎn)就是 Event - Time 處理的精髓，用來表示該收的數(shù)據(jù)已經(jīng)收到。

3.3 Watermarks

Flink 實(shí)際上是用 watermarks 來實(shí)現(xiàn) Event - Time 的功能。Watermarks 在 Flink 中也屬于特殊事件，其精髓在于當(dāng)某個(gè)運(yùn)算值收到帶有時(shí)間戳“ T ”的 watermarks 時(shí)就意味著它不會(huì)接收到新的數(shù)據(jù)了。使用 watermarks 的好處在于可以準(zhǔn)確預(yù)估收到數(shù)據(jù)的截止時(shí)間。舉例，假設(shè)預(yù)期收到數(shù)據(jù)時(shí)間與輸出結(jié)果時(shí)間的時(shí)間差延遲 5 分鐘，那么 Flink 中所有的 windows Operator 搜索 3 點(diǎn)至 4 點(diǎn)的數(shù)據(jù)，但因?yàn)榇嬖谘舆t需要再多等5分鐘直至收集完 4：05 分的數(shù)據(jù)，此時(shí)方能判定 4 點(diǎn)鐘的資料收集完成了，然后才會(huì)產(chǎn)出 3 點(diǎn)至 4 點(diǎn)的數(shù)據(jù)結(jié)果。這個(gè)時(shí)間段的結(jié)果對(duì)應(yīng)的就是 watermarks 的部分。

4.狀態(tài)保存與遷移

流式處理應(yīng)用無時(shí)無刻不在運(yùn)行，運(yùn)維上有幾個(gè)重要考量：

• 更改應(yīng)用邏輯/修 bug 等，如何將前一執(zhí)行的狀態(tài)遷移到新的執(zhí)行？

• 如何重新定義運(yùn)行的平行化程度？

• 如何升級(jí)運(yùn)算叢集的版本號(hào)？

Checkpoint 完美符合以上需求，不過 Flink 中還有另外一個(gè)名詞保存點(diǎn)（Savepoint），當(dāng)手動(dòng)產(chǎn)生一個(gè) Checkpoint 的時(shí)候，就叫做一個(gè) Savepoint。Savepoint 跟 Checkpoint 的差別在于檢查點(diǎn)是 Flink 對(duì)于一個(gè)有狀態(tài)應(yīng)用在運(yùn)行中利用分布式快照持續(xù)周期性的產(chǎn)生 Checkpoint，而 Savepoint 則是手動(dòng)產(chǎn)生的 Checkpoint，Savepoint 記錄著流式應(yīng)用中所有運(yùn)算元的狀態(tài)。

如圖，Savepoint A 和 Savepoint B，無論是變更底層代碼邏輯、修 bug 或是升級(jí) Flink 版本，重新定義應(yīng)用、計(jì)算的平行化程度等，最先需要做的事情就是產(chǎn)生 Savepoint。

Savepoint 產(chǎn)生的原理是在 Checkpoint barrier 流動(dòng)到所有的 Pipeline 中手動(dòng)插入從而產(chǎn)生分布式快照，這些分布式快照點(diǎn)即 Savepoint。Savepoint 可以放在任何位置保存，當(dāng)完成變更時(shí)，可以直接從 Savepoint 恢復(fù)、執(zhí)行。

從 Savepoint 的恢復(fù)執(zhí)行需要注意，在變更應(yīng)用的過程中時(shí)間在持續(xù)，如 Kafka 在持續(xù)收集資料，當(dāng)從 Savepoint 恢復(fù)時(shí)，Savepoint 保存著 Checkpoint 產(chǎn)生的時(shí)間以及 Kafka 的相應(yīng)位置，因此它需要恢復(fù)到最新的數(shù)據(jù)。無論是任何運(yùn)算，Event - Time 都可以確保產(chǎn)生的結(jié)果完全一致。

假設(shè)恢復(fù)后的重新運(yùn)算用 Process Event - Time，將 windows 窗口設(shè)為 1 小時(shí)，重新運(yùn)算能夠在 10 分鐘內(nèi)將所有的運(yùn)算結(jié)果都包含到單一的 windows 中。而如果使用 Event – Time，則類似于做 Bucketing。在 Bucketing 的狀況下，無論重新運(yùn)算的數(shù)量多大，最終重新運(yùn)算的時(shí)間以及 windows 產(chǎn)生的結(jié)果都一定能保證完全一致。

四、總結(jié)

本文首先從 Apache Flink 的定義、架構(gòu)、基本原理入手，對(duì)×××計(jì)算相關(guān)的基本概念進(jìn)行辨析，在此基礎(chǔ)上簡單回顧了大數(shù)據(jù)處理方式的歷史演進(jìn)以及有狀態(tài)的流式數(shù)據(jù)處理的原理，最后從目前有狀態(tài)的流式處理面臨的挑戰(zhàn)分析 Apache Flink 作為業(yè)界公認(rèn)為最好的流計(jì)算引擎之一所具備的天然優(yōu)勢。希望有助于大家厘清×××式處理引擎涉及的基本概念，能夠更加得心應(yīng)手的使用 Flink。

另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)scvps.cn，海內(nèi)外云服務(wù)器15元起步，三天無理由+7*72小時(shí)售后在線，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國服務(wù)器、虛擬主機(jī)、免備案服務(wù)器”等云主機(jī)租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡單易用、服務(wù)可用性高、性價(jià)比高”等特點(diǎn)與優(yōu)勢，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應(yīng)用場景需求。

當(dāng)前名稱：ApacheFlink零基礎(chǔ)入門（一）：基礎(chǔ)概念解析-創(chuàng)新互聯(lián)
文章網(wǎng)址：http://weahome.cn/article/ceogds.html