沈雅

Apache Flink持續(xù)保持高速發(fā)展，是Apache最活躍的社區(qū)之一。Flink 1.16共有240多個(gè)Contributor熱情參與，共完成了19個(gè)FLIP和1 100多個(gè)issue，給社區(qū)帶來非常多振奮人心的功能。

Flink已經(jīng)是流計(jì)算領(lǐng)域的領(lǐng)跑者，流批一體的概念逐漸得到大家的認(rèn)可，并在越來越多的公司成功落地。之前的流批一體更強(qiáng)調(diào)統(tǒng)一的API和統(tǒng)一的計(jì)算框架。2022年，在此基礎(chǔ)上，F(xiàn)link推出了Streaming Warehouse，進(jìn)一步升級了流批一體的概念：真正完成了流批一體的計(jì)算和流批一體的存儲(chǔ)的融合，從而實(shí)現(xiàn)流批一體的實(shí)時(shí)化分析。

在1.16版本里，F(xiàn)link社區(qū)對流、批都完成了眾多改進(jìn)：在批處理方面，完成了易用性、穩(wěn)定性、性能提升全方位的改進(jìn)，1.16是Fink批處理的里程碑式的版本，是走向成熟的重要一步。

易用性，引入SQL Gateway并完全兼容HiveServer2，用戶可以非常方便的提交Flink SQL作業(yè)和Hive SQL作業(yè)，同時(shí)也很容易連接到原有的Hive生態(tài)。

功能：Flink SQL用戶支持通過Join Hint指定Join策略，避免不合理的執(zhí)行計(jì)劃；Hive SQL的兼容性已經(jīng)達(dá)到94 %，用戶可以以極低的成本完成Hive到Flink的遷移。

穩(wěn)定性，通過預(yù)測執(zhí)行減少作業(yè)長尾以提高作業(yè)整體運(yùn)行穩(wěn)定性；支持自適應(yīng)Hash Join，通過失敗回滾機(jī)制避免作業(yè)失敗。

性能提升，對多分區(qū)表進(jìn)行動(dòng)態(tài)分區(qū)裁剪以提高處理效率，TPC-DS在10 TB規(guī)模數(shù)據(jù)集下性能提升了30 %；支持混合Shuffle模式，提高資源使用率和處理性能。

在流處理方面，也完成了很多重大改進(jìn)：

Changelog State Backend可以為用戶提供秒級甚至毫秒級Checkpoint，從而大幅提升容錯(cuò)體驗(yàn)，同時(shí)為事務(wù)性Sink作業(yè)提供更小的端到端延遲體驗(yàn)。

維表關(guān)聯(lián)在流處理中被廣泛被使用，引入了通用的緩存機(jī)制加快維表查詢速度，引入了可配置的異步模式提升維表查詢吞吐，引入可重試查詢機(jī)制解決維表延遲更新問題。這些功能都非常實(shí)用，解決了用戶經(jīng)常抱怨的痛點(diǎn)，支持了更豐富的場景。

從Flink SQL誕生第一天就存在一些非確定性操作，可能導(dǎo)致用戶作業(yè)出現(xiàn)錯(cuò)誤結(jié)果或作業(yè)運(yùn)行異常，這給用戶帶來了極大的困擾。

隨著流批一體的進(jìn)一步完善和Flink Table Store的不斷迭代，F(xiàn)link社區(qū)正一步一步推動(dòng)Streaming Warehouse從概念變?yōu)楝F(xiàn)實(shí)并走向成熟。

流式數(shù)倉

流式數(shù)倉（Streaming Warehouse）更準(zhǔn)確地說，其實(shí)是make data warehouse streaming，就是讓整個(gè)數(shù)倉所有分層的數(shù)據(jù)全部實(shí)時(shí)地流動(dòng)起來，從而實(shí)現(xiàn)一個(gè)具備端到端實(shí)時(shí)性的純流服務(wù)（Streaming Service），并且用一套統(tǒng)一API和計(jì)算框架來處理和分析所有流動(dòng)中的數(shù)據(jù)。

批處理

得益于在流處理的長期投資，流處理已經(jīng)成為流計(jì)算領(lǐng)域的領(lǐng)導(dǎo)者，在批處理上也投入了更多的精力，使其成為一個(gè)優(yōu)秀的批處理引擎。流批處理統(tǒng)一的整體體驗(yàn)也將會(huì)更加順暢。

SQL Gateway

從各個(gè)渠道反饋中了解到，SQL Gateway一直是用戶非常期待的功能，尤其是對批用戶。1.16里，該功能終于完成。SQL Gateway是對SQL Client的擴(kuò)展和增強(qiáng)，支持多租戶和插件式API協(xié)議（Endpoint），解決了SQL Client只能服務(wù)單用戶并且不能對接外部服務(wù)或組件的問題。當(dāng)前SQL Gateway已支持REST API和HiveServer2協(xié)議，用戶可以通過cURL、Postman以及各種編程語言的HTTP客戶端鏈接到SQL Gateway提交流作業(yè)、批作業(yè)，甚至OLAP作業(yè)。

Hive語法兼容

為了降低從Hive到Flink的遷移成本，這個(gè)版本里引入了HiveServer2協(xié)議并繼續(xù)改進(jìn)Hive語法的兼容性。

HiveServer2協(xié)議允許用戶使用Hive JDBC/Beeline和SQL Gateway進(jìn)行交互，Hive生態(tài)（DBeaver，Apache Superset，Apache DolphinScheduler，and Apache Zeppelin）也因此很容易遷移到Flink。當(dāng)用戶使用HiveServer2協(xié)議連接SQLGateway，SQL Gateway會(huì)自動(dòng)注冊Hive Catalog，自動(dòng)切換到Hive方言，自動(dòng)使用批處理模式提交作業(yè)，用戶可以得到和直接使用HiveServer2一樣的體驗(yàn)。

Hive語法已經(jīng)是大數(shù)據(jù)處理的事實(shí)標(biāo)準(zhǔn)，F(xiàn)link完善了對Hive語法的兼容，增加了對Hive若干生產(chǎn)中常用語法的支持。通過對Hive語法的兼容，可以幫助用戶將已有的Hive SQL任務(wù)遷移到Flink，并且方便熟悉Hive語法的用戶使用Hive語法編寫SQL以查詢注冊進(jìn)Flink中的表。到目前為止，基于Hive qtest測試集（包含12 000個(gè)SQL案例），Hive 2.3版本的查詢兼容性已達(dá)到94.1 %，如果排除ACID的查詢語句，則已達(dá)到97.3 %。

Join Hint

Hint一直是業(yè)界用來干預(yù)執(zhí)行計(jì)劃以改善優(yōu)化器缺點(diǎn)的通用解決方案。Join作為批作業(yè)中最廣泛使用的算子，F(xiàn)link支持多種Join策略。統(tǒng)計(jì)信息缺失或優(yōu)化器的代價(jià)模型不完善都會(huì)導(dǎo)致選出錯(cuò)誤Join策略，從而導(dǎo)致作業(yè)運(yùn)行慢甚至有運(yùn)行失敗的風(fēng)險(xiǎn)。用戶通過指定Join Hint，讓優(yōu)化器盡可能選擇用戶指定的Join策略，從而避免優(yōu)化器的各種不足，以確保批作業(yè)的生產(chǎn)可用性。

自適應(yīng)Hash Join

對于批作業(yè)而言，數(shù)據(jù)傾斜是非常常見的，而此時(shí)使用Hash Join可能運(yùn)行失敗，這是非常糟糕的體驗(yàn)。為了解決該問題，引入了自適應(yīng)的Hash Join：Join算子運(yùn)行時(shí)一旦Hash Join運(yùn)行失敗，可以自動(dòng)回退到Sort Merge Join，并且是Task粒度。通過該機(jī)制可確保Hash Join算子始終成功，從而提高了作業(yè)的穩(wěn)定性。

批處理的預(yù)測執(zhí)行

為了解決問題機(jī)器導(dǎo)致批作業(yè)處理慢的問題，F(xiàn)link 1.16引入了預(yù)測執(zhí)行。問題機(jī)器是指存在硬件問題、突發(fā)I/O繁忙或CPU負(fù)載高等問題的機(jī)器，這些問題可能會(huì)使得運(yùn)行在該機(jī)器上的任務(wù)比其他機(jī)器上的任務(wù)要慢得多，從而影響批處理作業(yè)的整體執(zhí)行時(shí)間。

當(dāng)啟用預(yù)測執(zhí)行時(shí)，F(xiàn)link將持續(xù)檢測慢任務(wù)。一旦檢測到慢任務(wù)，該任務(wù)所在的機(jī)器將被識(shí)別為問題機(jī)器，并通過黑名單機(jī)制被加黑。調(diào)度器將為慢任務(wù)創(chuàng)建新的執(zhí)行實(shí)例并將它們部署到未被加黑的節(jié)點(diǎn)，同時(shí)現(xiàn)有執(zhí)行實(shí)例也將繼續(xù)運(yùn)行。新的執(zhí)行實(shí)例和老的執(zhí)行實(shí)例將處理相同的輸入數(shù)據(jù)并產(chǎn)出相同的結(jié)果數(shù)據(jù)。一旦任何執(zhí)行實(shí)例率先完成，它將被視為該任務(wù)的唯一完成執(zhí)行實(shí)例，并且該任務(wù)的其余執(zhí)行實(shí)例都將被取消。

大多數(shù)現(xiàn)有Source都可以使用預(yù)測執(zhí)行。只有當(dāng)一個(gè)Source使用了SourceEvent時(shí)，它必須額外實(shí)現(xiàn)Supports Handle Execution Attempt Source Event接口以支持預(yù)測執(zhí)行。目前Sink尚不支持預(yù)測執(zhí)行，因此預(yù)測執(zhí)行不會(huì)在Sink上發(fā)生。

Web UI和REST API也有了改進(jìn)，以顯示任務(wù)的多個(gè)執(zhí)行實(shí)例和被加黑的TaskManager。

混合Shuffle模式

為批處理引入了一種新的混合Shuffle模式，它結(jié)合了Blocking Shuffle和Pipeline Shuffle（主要用于流式處理）的優(yōu)點(diǎn)。與Blocking Shuffle一樣，它不要求上下游任務(wù)同時(shí)運(yùn)行，這允許使用很少的資源執(zhí)行作業(yè)；與Pipeline Shuffle一樣，它不要求上游任務(wù)完成后才執(zhí)行下游任務(wù)，這在給定足夠資源情況下減少了作業(yè)的整體執(zhí)行時(shí)間。

用戶可以選擇不同的落盤策略，以滿足減少數(shù)據(jù)落盤或是降低任務(wù)重啟代價(jià)的不同需求。

注意：該功能為實(shí)驗(yàn)性的，并且默認(rèn)關(guān)閉。

Blocking Shuffle進(jìn)一步改進(jìn)

在這個(gè)版本中進(jìn)一步改進(jìn)了Blocking Shuffle的可用性和性能，包括自適應(yīng)網(wǎng)絡(luò)緩沖區(qū)分配、順序IO優(yōu)化和結(jié)果分區(qū)重用，允許多個(gè)消費(fèi)者節(jié)點(diǎn)重用同一個(gè)物理結(jié)果分區(qū)，以減少磁盤IO和存儲(chǔ)空間。在TPC-DS 10 TB規(guī)模的測試中，這些優(yōu)化可以實(shí)現(xiàn)7 %的整體性能提升。此外，還引入了2種壓縮率更高的壓縮算法（LZO和ZSTD）。與默認(rèn)的LZ4壓縮算法相比，可以進(jìn)一步減少存儲(chǔ)空間，但要付出一些CPU成本。

動(dòng)態(tài)分區(qū)裁剪

對于批作業(yè)，生產(chǎn)環(huán)境中分區(qū)表比非分區(qū)表使用更為廣泛。當(dāng)前Flink已經(jīng)支持靜態(tài)分區(qū)裁剪，即在優(yōu)化階段，優(yōu)化器將Filter中的Partition相關(guān)的過濾條件下推到Source Connector中從而減少不必要的分區(qū)讀取。星形模型是數(shù)據(jù)集市模式中最簡單且使用最廣泛的模式，很多用戶的作業(yè)沒法使用靜態(tài)分區(qū)裁剪，因?yàn)榉謪^(qū)裁剪信息在執(zhí)行時(shí)才能確定，這就需要?jiǎng)討B(tài)分區(qū)裁剪技術(shù)，即運(yùn)行時(shí)根據(jù)其他相關(guān)表的數(shù)據(jù)確定分區(qū)裁剪信息，從而減少對分區(qū)表中無效分區(qū)的讀取。通過TPC-DS 10 TB規(guī)模數(shù)據(jù)集的驗(yàn)證，該功能可提升30 %的性能。

流處理

在1.16中Checkpoint、SQL、Connector和其他領(lǐng)域都進(jìn)行了改進(jìn)，從而確保Flink在流計(jì)算領(lǐng)域繼續(xù)領(lǐng)先。

Changelog State Backend旨在令Checkpoint的間隔更短、更加可預(yù)測。這個(gè)版本在自身易用性上和與其他State Backend兼容性上做了諸多改進(jìn)，使其達(dá)到生產(chǎn)可用。

對于使用Flink構(gòu)建的云服務(wù)應(yīng)用來說，Rescaling是一種非常頻繁的操作。這個(gè)版本使用了RocksDB的區(qū)間刪除來優(yōu)化增量RocksDB State Backend的Rescaling性能。區(qū)間刪除被用來避免在Rescaling過程中大量的掃描和單點(diǎn)刪除操作，對有大量的狀態(tài)需要?jiǎng)h除的擴(kuò)并發(fā)來說，單個(gè)并發(fā)上的恢復(fù)速度可以提高2～10倍。

改善State Backend的監(jiān)測體驗(yàn)和可用性

這個(gè)版本還改善了狀態(tài)后臺(tái)的監(jiān)控體驗(yàn)和可用性。之前，RocksDB的日志位于它自己的DB目錄中，這使得調(diào)試RocksDB沒那么容易。這個(gè)版本讓RocksDB的日志默認(rèn)留在Flink的日志目錄中，新增了RocksDB相關(guān)的統(tǒng)計(jì)指標(biāo)，以幫助調(diào)試DB級別的性能，例如，在DB內(nèi)的總塊緩存命中/失敗計(jì)數(shù)。

支持透支緩沖區(qū)

透支緩沖區(qū)（Overdraft Buffers）旨在緩解反壓情況下Subtask被阻塞的概率，可以通過設(shè)置taskmanager.network. memory.max-overdraft-buffers-per-gate開啟。

從1.16開始，一個(gè)Flink的Subtask可以申請5個(gè)（默認(rèn)）額外的透支緩沖區(qū)。透支緩沖區(qū)會(huì)輕微地增加作業(yè)的內(nèi)存使用量，但可以極大地減少Checkpoint的間隔，特別是在開啟Unaligned Checkpoint情況下。只有當(dāng)前Subtask被下游Subtasks反壓且當(dāng)前Subtask需要請求超過1個(gè)網(wǎng)絡(luò)緩沖區(qū)（Network Buffer）才能完成當(dāng)前的操作時(shí)，透支緩沖區(qū)才會(huì)被使用。

對齊Checkpoint超時(shí)

這個(gè)版本更新了從Aligned Checkpoint（AC）切換到Unaligned Checkpoint（UC）的時(shí)間點(diǎn)。在開啟UC的情況下，如果配置了execution.checkpointing.aligned-checkpointtimeout，在啟動(dòng)時(shí)每個(gè)Checkpoint仍然是AC，但當(dāng)全局Checkpoint持續(xù)時(shí)間超過aligned-checkpoint-timeout時(shí)，如果AC還沒完成，那么Checkpoint將會(huì)轉(zhuǎn)換為UC。

以前，對一個(gè)Substask來說，AC到UC的切換需要等所有上游的Barriers到達(dá)后才能開始，在反壓嚴(yán)重的情況下，在checkpointing-timeout過期之前，下游的Substask可能無法完全地收到所有Barriers，從而導(dǎo)致Checkpoint失敗。

在這個(gè)版本中，如果上游Subtask中的Barrier無法在execution.checkpointing.aligned-checkpoint-timeout內(nèi)發(fā)送到下游，F(xiàn)link會(huì)讓上游的Subtask先切換成UC，以把Barrier發(fā)送到下游，從而減少反壓情況下Checkpoint超時(shí)的概率。

流計(jì)算的非確定性

Flink SQL用戶經(jīng)常抱怨理解流處理的成本太高，其中一個(gè)痛點(diǎn)是流處理中的非確定性（而且通常不直觀），它可能會(huì)導(dǎo)致錯(cuò)誤的結(jié)果或異常，而這些痛點(diǎn)在Flink SQL的早期就已經(jīng)存在了。

對于復(fù)雜的流作業(yè)，現(xiàn)在可以在運(yùn)行前檢測并解決潛在的正確性問題。如果問題不能完全解決，一個(gè)詳細(xì)的消息可以提示用戶如何調(diào)整SQL，以避免引入非確定性問題。

維表增強(qiáng)

維表關(guān)聯(lián)在流處理中被廣泛使用，在1.16中為此加入了多項(xiàng)優(yōu)化和增強(qiáng)：

支持通用的緩存機(jī)制和相關(guān)指標(biāo)，可以加速維表查詢；

通過作業(yè)配置或查詢提示支持可配置的異步模式（ALLOW_UNORDERED），在不影響正確性的前提下大大提升查詢吞吐；

可重試的查詢機(jī)制讓用戶解決維表數(shù)據(jù)更新延遲問題有了更多的手段。

異步I/O支持重試

為異步I/O引入了內(nèi)置的重試機(jī)制，它對用戶現(xiàn)有代碼是透明的，可以靈活地滿足用戶的重試和異常處理需求。

PyFlink

在Flink 1.15中引入了一種新的執(zhí)行模式：“線程”模式。在該模式下，用戶自定義的Python函數(shù)將通過JNI在JVM中執(zhí)行，而不是在獨(dú)立的Python進(jìn)程中執(zhí)行。但是，在Flink 1.15中，僅在Table API和SQL上的Python標(biāo)量函數(shù)的執(zhí)行上支持了該功能。在新版本中對該功能提供了更全面的支持，在Python DataStream API中以及在Table API和SQL的Python表值函數(shù)中，也支持了該功能。

除此之外，還補(bǔ)全Python API所缺失的最后幾處功能。在這個(gè)版本中，對Python DataStream API提供了更全面的支持，支持了旁路輸出、Broadcast State等功能，并完善了對于窗口功能的支持。在Python DataStream API中，添加了對于更多的Connector以及Format的支持，例如添加了對于Elasticsearch，Kinesis，Pulsar，Hybrid Source等Connector的支持以及對于Orc，Parquet等Format的支持。有了這些功能之后，Python API已經(jīng)基本對齊了Java和Scala API中絕大部分的重要功能，用戶已經(jīng)可以使用Python語言完成大多數(shù)類型Flink作業(yè)的開發(fā)。

DataStream中的緩存

支持通過DataStream#cache緩存Transformation的執(zhí)行結(jié)果。緩存的中間結(jié)果在首次計(jì)算中間結(jié)果時(shí)才生成，以便以后的作業(yè)可以重用該結(jié)果。如果緩存丟失，原始的Transformation將會(huì)被重新計(jì)算以得到結(jié)果。目前該功能只在批處理模式下支持。這個(gè)功能對于Python中的ML和交互式編程非常有用。

History Server及已完成作業(yè)的信息增強(qiáng)

在這個(gè)版本中加強(qiáng)了查看已完成作業(yè)的信息的體驗(yàn)。

JobManager / HistoryServer WebUI提供了詳細(xì)的執(zhí)行時(shí)間指標(biāo)，包括任務(wù)在每個(gè)執(zhí)行狀態(tài)下的耗時(shí)，以及在運(yùn)行過程中繁忙/空閑/反壓總時(shí)間。

Protobuf格式

Flink現(xiàn)在支持Protocol Buffers （Protobuf）格式，這允許直接在Table API或SQL應(yīng)用程序中使用這種格式。

為異步Sink引入可配置的RateLimitingStrategy

1.15中實(shí)現(xiàn)了異步Sink，允許用戶輕松實(shí)現(xiàn)自定義異步 Sink，新版本里此進(jìn)行擴(kuò)展以支持可配置的RateLimiting Strategy。這意味著Sink的實(shí)現(xiàn)者現(xiàn)在可以自定義其異步Sink在請求失敗時(shí)的行為方式，具體行為取決于特定的Sink。如果沒有指定RateLimitingStrategy，它將默認(rèn)使用AIMDScalingStrategy。