Flink 1.11 究竟有哪些易用性上的改善

本篇文章給大家分享的是有關(guān)Flink 1.11 究竟有哪些易用性上的改善，小編覺得挺實(shí)用的，因此分享給大家學(xué)習(xí)，希望大家閱讀完這篇文章后可以有所收獲，話不多說，跟著小編一起來看看吧。

• 首先在每個(gè)版本的規(guī)劃初期，會(huì)從志愿者中選出 1-2 名作為 Release Manager。1.11.0 版本我作為中國這邊的 Release Manager，同時(shí)還有一名來自 Ververica 的 Piotr Nowojski 作為德國方的 Release Manager，這在某種程度上也說明中國的開發(fā)者和貢獻(xiàn)度在整個(gè)社區(qū)的占比很重要。

• 接下來會(huì)進(jìn)行這個(gè)版本的 Feature Kickoff。在一些大的方向上，社區(qū)的規(guī)劃周期可能比較久，會(huì)分階段、分步驟跨越多個(gè)版本完成，確保質(zhì)量。每個(gè)版本的側(cè)重點(diǎn)也會(huì)有所不同，比如前兩個(gè)版本側(cè)重于批處理的加強(qiáng)，而這個(gè)版本更側(cè)重于流處理易用性的提升。社區(qū)規(guī)劃的 Feature 列表會(huì)在郵件列表中發(fā)起討論，以收集更多的用戶/開發(fā)者意見和反饋。

• 一般的開發(fā)周期為 2-3 個(gè)月時(shí)間，提前會(huì)明確規(guī)劃出大概的 Feature Freeze 時(shí)間，之后進(jìn)行 Release Candidate 的發(fā)布和測試、以及 Bug Fix。一般經(jīng)過幾輪的迭代周期后會(huì)正式投票通過一個(gè)相對穩(wěn)定的 Candidate 版本，然后基于這個(gè)版本正式發(fā)布。

CREATE TABLE my_table (  ...) WITH (  'connector'='...', -- e.g. 'kafka'  'format'='debezium-json',  'debezium-json.schema-include'='true' -- default: false (Debezium can be configured to include or exclude the message schema)  'debezium-json.ignore-parse-errors'='true' -- default: false);

• Hive 寫入：FLIP-115 完善擴(kuò)展了 FileSystem Connector 的基礎(chǔ)能力和實(shí)現(xiàn)，Table/SQL 層的 sink 可以支持各種格式（CSV、Json、Avro、Parquet、ORC），而且支持 Hive Table 的所有格式。

• Partition 支持：數(shù)據(jù)導(dǎo)入 Hive 引入 Partition 提交機(jī)制來控制可見性，通過sink.partition-commit.trigger 控制 Partition 提交的時(shí)機(jī)，通過 sink.partition-commit.policy.kind 選擇提交策略，支持 SUCCESS 文件和 Metastore 提交。

• Hive 讀?。簩?shí)時(shí)化的流式讀取 Hive，通過監(jiān)控 Partition 生成增量讀取新 Partition，或者監(jiān)控文件夾內(nèi)新文件生成來增量讀取新文件。

• FLIP-123 通過 Hive Dialect 為用戶提供語法兼容，這樣用戶無需在 Flink 和 Hive 的 CLI 之間切換，可以直接遷移 Hive 腳本到 Flink 中執(zhí)行。

• 提供 Hive 相關(guān)依賴的內(nèi)置支持，避免用戶自己下載所需的相關(guān)依賴?，F(xiàn)在只需要單獨(dú)下載一個(gè)包，配置 HADOOP_CLASSPATH 就可以運(yùn)行。

• 對用戶而言，在 Flink 中改造已有的 Source 或者重新實(shí)現(xiàn)一個(gè)生產(chǎn)級的 Source Connector 不是一件容易的事情，具體體現(xiàn)在沒有公共的代碼可以復(fù)用，而且需要理解很多 Flink 內(nèi)部細(xì)節(jié)以及實(shí)現(xiàn)具體的 Event Time 分配、Watermark 產(chǎn)出、Idleness 監(jiān)測、線程模型等。

• 批和流的場景需要實(shí)現(xiàn)不同的 Source。

• Partitions/Splits/Shards 概念在接口中沒有顯式表達(dá)，比如 Split 的發(fā)現(xiàn)邏輯和數(shù)據(jù)消費(fèi)都耦合在 Source Sunction 的實(shí)現(xiàn)中，這樣在實(shí)現(xiàn) Kafka 或 Kinesis 類型的 Source 時(shí)增加了復(fù)雜性。

• 在 Runtime 執(zhí)行層，Checkpoint 鎖被 Source Function 搶占會(huì)帶來一系列問題，框架很難進(jìn)行優(yōu)化。

• 首先在 Job Manager 和 Task Manager 中分別引入兩種不同的組件 Split Enumerator 和 Source Reader，解耦 Split 發(fā)現(xiàn)和對應(yīng)的消費(fèi)處理，同時(shí)方便隨意組合不同的策略。比如現(xiàn)有的 Kafka Connector 中有多種不同的 Partition 發(fā)現(xiàn)策略和實(shí)現(xiàn)耦合在一起，在新的架構(gòu)下，我們只需要實(shí)現(xiàn)一種 Source Reader，就可以適配多種 Split Enumerator 的實(shí)現(xiàn)來對應(yīng)不同的 Partition 發(fā)現(xiàn)策略。

• 在新架構(gòu)下實(shí)現(xiàn)的 Source Connector 可以做到批流統(tǒng)一，唯一的小區(qū)別是對批場景的有限輸入，Split Enumerator 會(huì)產(chǎn)出固定數(shù)量的 Split 集合并且每個(gè) Split 都是有限數(shù)據(jù)集；對于流場景的無限輸入，Split Enumerator 要么產(chǎn)出無限多的 Split 或者 Split 自身是無限數(shù)據(jù)集。

• 復(fù)雜的 Timestamp Assigner 以及 Watermark Generator 透明的內(nèi)置在 Source Reader 模塊內(nèi)運(yùn)行，對用戶來說是無感知的。這樣用戶如果想實(shí)現(xiàn)新的 Source Connector，一般不再需要重復(fù)實(shí)現(xiàn)這部分功能。

• 每次調(diào)用可以處理 N 條數(shù)據(jù)。

• 數(shù)據(jù)格式基于 Apache Arrow，大大降低了 Java、Python 進(jìn)程之間的序列化/反序列化開銷。

• 方便 Python 用戶基于 Numpy 和 Pandas 等數(shù)據(jù)分析領(lǐng)域常用的 Python 庫，開發(fā)高性能的 Python UDF。

• PyFlink table 和 Pandas DataFrame 之間無縫切換（FLIP-120），增強(qiáng) Pandas 生態(tài)的易用性和兼容性。

• Table & SQL 中可以定義和使用 Python UDTF（FLINK-14500），不再必需 Java/Scala UDTF。

• Cython 優(yōu)化 Python UDF 的性能（FLIP-121），對比 1.10.0 可以提升 30 倍。

• Python UDF 中用戶自定義 Metric（FLIP-112），方便監(jiān)控和調(diào)試 UDF 的執(zhí)行。

• Session 模式：提前啟動(dòng)一個(gè)集群，所有作業(yè)都共享這個(gè)集群的資源運(yùn)行。優(yōu)勢是避免每個(gè)作業(yè)單獨(dú)啟動(dòng)集群帶來的額外開銷，缺點(diǎn)是隔離性稍差。如果一個(gè)作業(yè)把某個(gè) Task Manager（TM）容器搞掛，會(huì)導(dǎo)致這個(gè)容器內(nèi)的所有作業(yè)都跟著重啟。雖然每個(gè)作業(yè)有自己獨(dú)立的 Job Manager（JM）來管理，但是這些 JM 都運(yùn)行在一個(gè)進(jìn)程中，容易帶來負(fù)載上的瓶頸。

• Per-job 模式：為了解決 Session 模式隔離性差的問題，每個(gè)作業(yè)根據(jù)資源需求啟動(dòng)獨(dú)立的集群，每個(gè)作業(yè)的 JM 也是運(yùn)行在獨(dú)立的進(jìn)程中，負(fù)載相對小很多。

• 上游輸出端控制單個(gè) Sub Partition 堆積 Buffer 的最大閾值（Backlog），避免負(fù)載不均場景下單個(gè)鏈路上堆積大量 Buffer。

• 在不影響網(wǎng)絡(luò)吞吐性能的情況下合理修改上下游默認(rèn)的 Buffer 配置。

• 上下游數(shù)據(jù)傳輸?shù)幕A(chǔ)協(xié)議進(jìn)行了調(diào)整，允許單個(gè)數(shù)據(jù)鏈路可以配置 0 個(gè)獨(dú)占 Buffer 而不死鎖，這樣總的 Buffer 數(shù)量和作業(yè)并發(fā)規(guī)模解耦。根據(jù)實(shí)際需求在吞吐性能和 Checkpoint 速度兩者之間權(quán)衡，自定義 Buffer 配比。

• Checkpoint Barrier 跨數(shù)據(jù) Buffer 傳輸，不在輸入輸出隊(duì)列排隊(duì)等待處理，這樣就和算子的計(jì)算能力解耦，Barrier 在節(jié)點(diǎn)之間的傳輸只有網(wǎng)絡(luò)延時(shí)，可以忽略不計(jì)。

• 每個(gè)算子多個(gè)輸入鏈路之間不需要等待 Barrier 對齊來執(zhí)行 Checkpoint，第一個(gè)到的 Barrier 就可以提前觸發(fā) Checkpoint，這樣可以進(jìn)一步提速 Checkpoint，不會(huì)因?yàn)閭€(gè)別鏈路的延遲而影響整體。

• 為了和之前 Aligned Checkpoint 的語義保持一致，所有未被處理的輸入輸出數(shù)據(jù) Buffer 都將作為 Channel State 在 Checkpoint 執(zhí)行時(shí)進(jìn)行快照持久化，在 Failover 時(shí)連同 Operator State 一同進(jìn)行恢復(fù)。換句話說，Aligned 機(jī)制保證的是 Barrier 前面所有數(shù)據(jù)必須被處理完，狀態(tài)實(shí)時(shí)體現(xiàn)到 Operator State 中；而 Unaligned 機(jī)制把 Barrier 前面的未處理數(shù)據(jù)所反映的 Operator State 延后到 Failover Restart 時(shí)通過 Channel State 回放進(jìn)行體現(xiàn)，從狀態(tài)恢復(fù)的角度來說最終都是一致的。注意這里雖然引入了額外的 In-Flight Buffer 的持久化，但是這個(gè)過程實(shí)際是在 Checkpoint 的異步階段完成的，同步階段只是進(jìn)行了輕量級的 Buffer 引用，所以不會(huì)過多占用算子的計(jì)算時(shí)間而影響吞吐性能。

  

以上就是Flink 1.11 究竟有哪些易用性上的改善，小編相信有部分知識(shí)點(diǎn)可能是我們?nèi)粘９ぷ鲿?huì)見到或用到的。希望你能通過這篇文章學(xué)到更多知識(shí)。更多詳情敬請關(guān)注億速云行業(yè)資訊頻道。