溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
作者：陳守元、戴資力
Apache Flink 是一個分布式大數據處理引擎，可對有限數據流和無限數據流進行有狀態或無狀態的計算，能夠部署在各種集群環境，對各種規模大小的數據進行快速計算。
了解 Flink 應用開發需要先理解 Flink 的 Streams、State、Time 等基礎處理語義以及 Flink 兼顧靈活性和方便性的多層次 API。
Streams：流，分為有限數據流與無限數據流，unbounded stream 是有始無終的數據流，即無限數據流；而 bounded stream 是限定大小的有始有終的數據集合，即有限數據流，二者的區別在于無限數據流的數據會隨時間的推演而持續增加，計算持續進行且不存在結束的狀態，相對的有限數據流數據大小固定，計算最終會完成并處于結束的狀態。
State，狀態是計算過程中的數據信息，在容錯恢復和 Checkpoint 中有重要的作用，流計算在本質上是 Incremental Processing，因此需要不斷查詢保持狀態；另外，為了確保 Exactly- once 語義，需要數據能夠寫入到狀態中；而持久化存儲，能夠保證在整個分布式系統運行失敗或者掛掉的情況下做到 Exactly- once，這是狀態的另外一個價值。
Time，分為 Event time、Ingestion time、Processing time，Flink 的無限數據流是一個持續的過程，時間是我們判斷業務狀態是否滯后，數據處理是否及時的重要依據。
在架構部分，主要分為以下四點：
cdn.xitu.io/2019/7/2/16bb2c32e9736d17?w=960&h=540&f=jpeg&s=98000">
第一， Flink 具備統一的框架處理有界和×××兩種數據流的能力
第二， 部署靈活，Flink 底層支持多種資源調度器，包括 Yarn、Kubernetes 等。Flink 自身帶的 Standalone 的調度器，在部署上也十分靈活。
第三， 極高的可伸縮性，可伸縮性對于分布式系統十分重要，阿里巴巴雙11大屏采用 Flink 處理海量數據，使用過程中測得 Flink 峰值可達 17 億/秒。
第四， 極致的流式處理性能。Flink 相對于 Storm 最大的特點是將狀態語義完全抽象到框架中，支持本地狀態讀取，避免了大量網絡 IO，可以極大提升狀態存取的性能。
后面會有專門課程講解，此處簡單分享 Flink 關于運維及業務監控的內容：
Flink 具備 7 X 24 小時高可用的 SOA（面向服務架構），原因是在實現上 Flink 提供了一致性的 Checkpoint。Checkpoint 是 Flink 實現容錯機制的核心，它周期性的記錄計算過程中 Operator 的狀態，并生成快照持久化存儲。當 Flink 作業發生故障崩潰時，可以有選擇的從 Checkpoint 中恢復，保證了計算的一致性。
Data Pipeline 的核心場景類似于數據搬運并在搬運的過程中進行部分數據清洗或者處理，而整個業務架構圖的左邊是 Periodic ETL，它提供了流式 ETL 或者實時 ETL，能夠訂閱消息隊列的消息并進行處理，清洗完成后實時寫入到下游的 Database 或 File system 中。場景舉例：
當下游要構建實時數倉時，上游則可能需要實時的 Stream ETL。這個過程會進行實時清洗或擴展數據，清洗完成后寫入到下游的實時數倉的整個鏈路中，可保證數據查詢的時效性，形成實時數據采集、實時數據處理以及下游的實時 Query。
搜索引擎這塊以淘寶為例，當賣家上線新商品時，后臺會實時產生消息流，該消息流經過 Flink 系統時會進行數據的處理、擴展。然后將處理及擴展后的數據生成實時索引，寫入到搜索引擎中。這樣當淘寶賣家上線新商品時，能在秒級或者分鐘級實現搜索引擎的搜索。
Data Analytics，如圖，左邊是 Batch Analytics，右邊是 Streaming Analytics。Batch Analysis 就是傳統意義上使用類似于 Map Reduce、Hive、Spark Batch 等，對作業進行分析、處理、生成離線報表，Streaming Analytics 使用流式分析引擎如 Storm，Flink 實時處理分析數據，應用較多的場景如實時大屏、實時報表。
從某種程度上來說，所有的實時的數據處理或者是流式數據處理都是屬于 Data Driven，流計算本質上是 Data Driven 計算。應用較多的如風控系統，當風控系統需要處理各種各樣復雜的規則時，Data Driven 就會把處理的規則和邏輯寫入到Datastream 的 API 或者是 ProcessFunction 的 API 中，然后將邏輯抽象到整個 Flink 引擎中，當外面的數據流或者是事件進入就會觸發相應的規則，這就是 Data Driven 的原理。在觸發某些規則后，Data Driven 會進行處理或者是進行預警，這些預警會發到下游產生業務通知，這是 Data Driven 的應用場景，Data Driven 在應用上更多應用于復雜事件的處理。
傳統批處理方法是持續收取數據，以時間作為劃分多個批次的依據，再周期性地執行批次運算。但假設需要計算每小時出現事件轉換的次數，如果事件轉換跨越了所定義的時間劃分，傳統批處理會將中介運算結果帶到下一個批次進行計算；除此之外，當出現接收到的事件順序顛倒情況下，傳統批處理仍會將中介狀態帶到下一批次的運算結果中，這種處理方式也不盡如人意。
第一點，要有理想方法，這個理想方法是引擎必須要有能力可以累積狀態和維護狀態，累積狀態代表著過去歷史中接收過的所有事件，會影響到輸出。
第二點，時間，時間意味著引擎對于數據完整性有機制可以操控，當所有數據都完全接受到后，輸出計算結果。
第三點，理想方法模型需要實時產生結果，但更重要的是采用新的持續性數據處理模型來處理實時數據，這樣才最符合 continuous data 的特性。
流式處理簡單來講即有一個無窮無盡的數據源在持續收取數據，以代碼作為數據處理的基礎邏輯，數據源的數據經過代碼處理后產生出結果，然后輸出，這就是流式處理的基本原理。
假設 Input Streams 有很多個使用者，每個使用者都有自己的 ID，如果計算每個使用者出現的次數，我們需要讓同一個使用者的出現事件流到同一運算代碼，這跟其他批次需要做 group by 是同樣的概念，所以跟 Stream 一樣需要做分區，設定相應的 key，然后讓同樣的 key 流到同一個 computation instance 做同樣的運算。
如圖，上述代碼中定義了變數 X，X 在數據處理過程中會進行讀和寫，在最后輸出結果時，可以依據變數 X 決定輸出的內容，即狀態 X 會影響最終的輸出結果。這個過程中，第一個重點是先進行了狀態 co-partitioned key by，同樣的 key 都會流到 computation instance，與使用者出現次數的原理相同，次數即所謂的狀態，這個狀態一定會跟同一個 key 的事件累積在同一個 computation instance。
相當于根據輸入流的 key 重新分區的 狀態，當分區進入 stream 之后，這個 stream 會累積起來的狀態也變成 copartiton 了。第二個重點是 embeded local state backend。有狀態分散式流式處理的引擎，狀態可能會累積到非常大，當 key 非常多時，狀態可能就會超出單一節點的 memory 的負荷量，這時候狀態必須有狀態后端去維護它；在這個狀態后端在正常狀況下，用 in-memory 維護即可。
當我們考慮狀態容錯時難免會想到精確一次的狀態容錯，應用在運算時累積的狀態，每筆輸入的事件反映到狀態，更改狀態都是精確一次，如果修改超過一次的話也意味著數據引擎產生的結果是不可靠的。
如何確保狀態擁有精確一次（Exactly-once guarantee）的容錯保證？
如何在分散式場景下替多個擁有本地狀態的運算子產生一個全域一致的快照（Global consistent snapshot）？
還是以使用者出現次數來看，如果某個使用者出現的次數計算不準確，不是精確一次，那么產生的結果是無法作為參考的。在考慮精確的容錯保證前，我們先考慮最簡單的使用場景，如無限流的數據進入，后面單一的 Process 進行運算，每處理完一筆計算即會累積一次狀態，這種情況下如果要確保 Process 產生精確一次的狀態容錯，每處理完一筆數據，更改完狀態后進行一次快照，快照包含在隊列中并與相應的狀態進行對比，完成一致的快照，就能確保精確一次。
Flink 作為分布式的處理引擎，在分布式的場景下，進行多個本地狀態的運算，只產生一個全域一致的快照，如需要在不中斷運算值的前提下產生全域一致的快照，就涉及到分散式狀態容錯。
關于 Global consistent snapshot，當 Operator 在分布式的環境中，在各個節點做運算，首先產生 Global consistent snapshot 的方式就是處理每一筆數據的快照點是連續的，這筆運算流過所有的運算值，更改完所有的運算值后，能夠看到每一個運算值的狀態與該筆運算的位置，即可稱為 consistent snapshot，當然，Global consistent snapshot 也是簡易場景的延伸。
首先了解一下 Checkpoint，上面提到連續性快照每個 Operator 運算值本地的狀態后端都要維護狀態，也就是每次將產生檢查點時會將它們傳入共享的 DFS 中。當任何一個 Process 掛掉后，可以直接從三個完整的 Checkpoint 將所有的運算值的狀態恢復，重新設定到相應位置。Checkpoint 的存在使整個 Process 能夠實現分散式環境中的 Exactly-once。
關于 Flink 如何在不中斷運算的狀況下持續產生 Global consistent snapshot，其方式是基于用 simple lamport 演算法機制下延伸的。已知的一個點 Checkpoint barrier， Flink 在某個 Datastream 中會一直安插 Checkpoint barrier，Checkpoint barrier 也會 N — 1等等，Checkpoint barrier N 代表著所有在這個范圍里面的數據都是Checkpoint barrier N。
舉例：假設現在需要產生 Checkpoint barrier N，但實際上在 Flink 中是由 job manager 觸發 Checkpoint，Checkpoint 被觸發后開始從數據源產生 Checkpoint barrier。當 job 開始做 Checkpoint barrier N 的時候，可以理解為 Checkpoint barrier N 需要逐步填充左下角的表格。
如圖，當部分事件標為紅色，Checkpoint barrier N 也是紅色時，代表著這些數據或事件都由 Checkpoint barrier N 負責。Checkpoint barrier N 后面白色部分的數據或事件則不屬于 Checkpoint barrier N。
在以上的基礎上，當數據源收到 Checkpoint barrier N 之后會先將自己的狀態保存，以讀取 Kafka 資料為例，數據源的狀態就是目前它在 Kafka 分區的位置，這個狀態也會寫入到上面提到的表格中。下游的 Operator 1 會開始運算屬于 Checkpoint barrier N 的數據，當 Checkpoint barrier N 跟著這些數據流動到 Operator 1 之后,Operator 1 也將屬于 Checkpoint barrier N 的所有數據都反映在狀態中，當收到 Checkpoint barrier N 時也會直接對 Checkpoint 去做快照。
當快照完成后繼續往下游走，Operator 2 也會接收到所有數據，然后搜索 Checkpoint barrier N 的數據并直接反映到狀態，當狀態收到 Checkpoint barrier N 之后也會直接寫入到 Checkpoint N 中。以上過程到此可以看到 Checkpoint barrier N 已經完成了一個完整的表格，這個表格叫做 Distributed Snapshots，即分布式快照。分布式快照可以用來做狀態容錯，任何一個節點掛掉的時候可以在之前的 Checkpoint 中將其恢復。繼續以上 Process，當多個 Checkpoint 同時進行，Checkpoint barrier N 已經流到 job manager 2，Flink job manager 可以觸發其他的 Checkpoint，比如 Checkpoint N + 1，Checkpoint N + 2 等等也同步進行，利用這種機制，可以在不阻擋運算的狀況下持續地產生 Checkpoint。
狀態維護即用一段代碼在本地維護狀態值，當狀態值非常大時需要本地的狀態后端來支持。
如圖，在 Flink 程序中，可以采用 getRuntimeContext().getState(desc); 這組 API 去注冊狀態。Flink 有多種狀態后端，采用 API 注冊狀態后，讀取狀態時都是通過狀態后端來讀取的。Flink 有兩種不同的狀態值，也有兩種不同的狀態后端：
Flink 目前支持以上兩種狀態后端，一種是純 memory 的狀態后端，另一種是有資源磁盤的狀態后端，在維護狀態時可以根據狀態的數量選擇相應的狀態后端。
3.1 不同時間種類
在 Flink 及其他進階的流式處理引擎出現之前，大數據處理引擎一直只支持 Processing-time 的處理。假設定義一個運算 windows 的窗口，windows 運算設定每小時進行結算。以 Processing-time 進行運算時可以發現數據引擎將 3 點至 4 點間收到的數據做結算。實際上在做報表或者分析結果時是想了解真實世界中 3 點至 4 點之間實際產生數據的輸出結果，了解實際數據的輸出結果就必須采用 Event – Time 了。
如圖，Event - Time 相當于事件，它在數據最源頭產生時帶有時間戳，后面都需要用時間戳來進行運算。用圖來表示，最開始的隊列收到數據，每小時對數據劃分一個批次，這就是 Event - Time Process 在做的事情。
3.2 Event - Time 處理
Event - Time 是用事件真實產生的時間戳去做 Re-bucketing，把對應時間 3 點到 4 點的數據放在 3 點到 4 點的 Bucket，然后 Bucket 產生結果。所以 Event - Time 跟 Processing - time 的概念是這樣對比的存在。
Event - Time 的重要性在于記錄引擎輸出運算結果的時間。簡單來說，流式引擎連續 24 小時在運行、搜集資料，假設 Pipeline 里有一個 windows Operator 正在做運算，每小時能產生結果，何時輸出 windows 的運算值，這個時間點就是 Event - Time 處理的精髓，用來表示該收的數據已經收到。
3.3 Watermarks
Flink 實際上是用 watermarks 來實現 Event - Time 的功能。Watermarks 在 Flink 中也屬于特殊事件，其精髓在于當某個運算值收到帶有時間戳“ T ”的 watermarks 時就意味著它不會接收到新的數據了。使用 watermarks 的好處在于可以準確預估收到數據的截止時間。舉例，假設預期收到數據時間與輸出結果時間的時間差延遲 5 分鐘，那么 Flink 中所有的 windows Operator 搜索 3 點至 4 點的數據，但因為存在延遲需要再多等5分鐘直至收集完 4：05 分的數據，此時方能判定 4 點鐘的資料收集完成了，然后才會產出 3 點至 4 點的數據結果。這個時間段的結果對應的就是 watermarks 的部分。
流式處理應用無時無刻不在運行，運維上有幾個重要考量：
更改應用邏輯/修 bug 等，如何將前一執行的狀態遷移到新的執行？
如何重新定義運行的平行化程度？
Checkpoint 完美符合以上需求，不過 Flink 中還有另外一個名詞保存點（Savepoint），當手動產生一個 Checkpoint 的時候，就叫做一個 Savepoint。Savepoint 跟 Checkpoint 的差別在于檢查點是 Flink 對于一個有狀態應用在運行中利用分布式快照持續周期性的產生 Checkpoint，而 Savepoint 則是手動產生的 Checkpoint，Savepoint 記錄著流式應用中所有運算元的狀態。
如圖，Savepoint A 和 Savepoint B，無論是變更底層代碼邏輯、修 bug 或是升級 Flink 版本，重新定義應用、計算的平行化程度等，最先需要做的事情就是產生 Savepoint。
Savepoint 產生的原理是在 Checkpoint barrier 流動到所有的 Pipeline 中手動插入從而產生分布式快照，這些分布式快照點即 Savepoint。Savepoint 可以放在任何位置保存，當完成變更時，可以直接從 Savepoint 恢復、執行。
從 Savepoint 的恢復執行需要注意，在變更應用的過程中時間在持續，如 Kafka 在持續收集資料，當從 Savepoint 恢復時，Savepoint 保存著 Checkpoint 產生的時間以及 Kafka 的相應位置，因此它需要恢復到最新的數據。無論是任何運算，Event - Time 都可以確保產生的結果完全一致。
假設恢復后的重新運算用 Process Event - Time，將 windows 窗口設為 1 小時，重新運算能夠在 10 分鐘內將所有的運算結果都包含到單一的 windows 中。而如果使用 Event – Time，則類似于做 Bucketing。在 Bucketing 的狀況下，無論重新運算的數量多大，最終重新運算的時間以及 windows 產生的結果都一定能保證完全一致。
本文首先從 Apache Flink 的定義、架構、基本原理入手，對×××計算相關的基本概念進行辨析，在此基礎上簡單回顧了大數據處理方式的歷史演進以及有狀態的流式數據處理的原理，最后從目前有狀態的流式處理面臨的挑戰分析 Apache Flink 作為業界公認為最好的流計算引擎之一所具備的天然優勢。希望有助于大家厘清×××式處理引擎涉及的基本概念，能夠更加得心應手的使用 Flink。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
