官宣｜Apache Flink 1.13.0 正式釋出，流處理應用更加簡單高效

Flink 1.13 釋出了！Flink 1.13 包括了超過 200 名貢獻者所提交的 1000 多項修復和優化。

這一版本中，Flink 的一個主要目標取得了重要進展，即讓流處理應用的使用和普通應用一樣簡單和自然。Flink 1.13 新引入的被動擴縮容使得流作業的擴縮容和其它應用一樣簡單，使用者僅需要修改併發度即可。

這個版本還包括一系列重要改動使使用者可以更好的理解流作業的效能。當流作業的效能不及預期的時候，這些改動可以使使用者可以更好的分析原因。這些改動包括用於識別瓶頸節點的負載和反壓視覺化、分析運算元熱點程式碼的 CPU 火焰圖和分析 State Backend 狀態的 State 訪問效能指標。

除了這些特性外，Flink 社群還添加了大量的其它優化，我們會在本文後續討論其中的一些。我們希望使用者可以享受新的版本和特性帶來的便利，在本文最後，我們還會介紹升級Flink版本需要注意的一些變化。

我們鼓勵使用者下載試用新版 Flink 並且通過郵件列表和 JIRA 來反饋遇到的問題。

重要特性

被動擴縮容

Flink 項目的一個初始目標，就是希望流處理應用可以像普通應用一樣簡單和自然，被動擴縮容是 Flink 針對這一目標上的最新進展。

當考慮資源管理和部分的時候，Flink 有兩種可能的模式。使用者可以將 Flink 應用部署到 k8s、yarn 等資源管理系統之上，並且由 Flink 主動的來管理資源並按需分配和釋放資源。這一模式對於經常改變資源需求的作業和應用非常有用，比如批作業和實時 SQL 查詢。在這種模式下，Flink 所啟動的 Worker 數量是由應用設定的併發度決定的。在 Flink 中我們將這一模式叫做主動擴縮容。

對於長時間運行的流處理應用，一種更適合的模型是使用者只需要將作業像其它的長期運行的服務一樣啟動起來，而不需要考慮是部署在 k8s、yarn 還是其它的資源管理平臺上，並且不需要考慮需要申請的資源的數量。相反，它的規模是由所分配的 worker 數量來決定的。當 worker 數量發生變化時，Flink 自動的改動應用的併發度。在 Flink 中我們將這一模式叫做被動擴縮容。

Flink 的 Application 部署模式開啟了使 Flink 作業更接近普通應用（即啟動 Flink 作業不需要執行兩個獨立的步驟來啟動叢集和提交應用）的努力，而被動擴縮容完成了這一目標：使用者不再需要使用額外的工具（如指令碼、K8s 運算元）來讓 worker 的數量與應用併發度設定保持一致。

使用者現在可以將自動擴縮容的工具應用到 Flink 應用之上，就像普通的應用程式一樣，只要使用者瞭解擴縮容的代價：有狀態的流應用在擴縮容的時候需要將狀態重新分發。

如果想要嘗試被動擴縮容，使用者可以增加 scheduler-mode: reactive 這一配置項，然後啟動一個應用叢集（Standalone 或者 K8s）。更多細節見被動擴縮容的文件。

分析應用的效能

對所有應用程式來說，能夠簡單的分析和理解應用的效能是非常關鍵的功能。這一功能對 Flink 更加重要，因為 Flink 應用一般是資料密集的（即需要處理大量的資料）並且需要在（近）實時的延遲內給出結果。

當 Flink 應用處理的速度跟不上資料輸入的速度時，或者當一個應用佔用的資源超過預期，下文介紹的這些工具可以幫你分析原因。

瓶頸檢測與反壓監控

Flink 效能分析首先要解決的問題經常是：哪個運算元是瓶頸？

為了回答這一問題，Flink 引入了描述作業繁忙（即在處理資料）與反壓（由於下游運算元不能及時處理結果而無法繼續輸出）程度的指標。應用中可能的瓶頸是那些繁忙併且上游被反壓的運算元。

Flink 1.13 優化了反壓檢測的邏輯（使用基於任務 Mailbox 計時，而不在再於堆棧取樣），並且重新實現了作業圖的 UI 展示：Flink 現在在 UI 上通過顏色和數值來展示繁忙和反壓的程度。

Web UI 中的 CPU 火焰圖

Flink 關於效能另一個經常需要回答的問題：瓶頸運算元中的哪部分計算邏輯消耗巨大？

針對這一問題，一個有效的視覺化工具是火焰圖。它可以幫助回答以下問題：

哪個方法調現在在佔用 CPU？不同方法佔用 CPU 的比例如何？一個方法被呼叫的棧是什麼樣子的？

火焰圖是通過重複取樣執行緒的堆棧來構建的。在火焰圖中，每個方法呼叫被表示為一個矩形，矩形的長度與這個方法出現在取樣中的次數成正比。火焰圖在 UI 上的一個例子如下圖所示。

火焰圖的文件包括啟用這一功能的更多細節和指令。

State 訪問延遲指標

另一個可能的效能瓶頸是 state backend，尤其是當作業的 state 超過記憶體容量而必須使用 RocksDB state backend 時。

這裡並不是想說 RocksDB 效能不夠好（我們非常喜歡 RocksDB！），但是它需要滿足一些條件才能達到最好的效能。例如，使用者可能很容易遇到非故意的在雲上由於使用了錯誤的磁碟資源類型而不能滿足 RockDB 的 IO 效能需求的問題。

基於 CPU 火焰圖，新的 State Backend 的延遲指標可以幫助使用者更好的判斷效能不符合預期是否是由 State Backend 導致的。例如，如果使用者發現 RocksDB 的單次訪問需要幾毫秒的時間，那麼就需要檢視記憶體和 I/O 的配置。這些指標可以通過設定 state.backend.rocksdb.latency-track-enabled 這一選項來啟用。這些指標是通過取樣的方式來監控效能的，所以它們對 RocksDB State Backend 的效能影響是微不足道的。

通過 Savepoint 來切換 State Backend

使用者現在可以在從一個 Savepoint 重啟時切換一個 Flink 應用的 State Backend。這使得 Flink 應用不再被限制只能使用應用首次運行時選擇的 State Backend。

基於這一功能，使用者現在可以首先使用一個 HashMap State Backend（純記憶體的 State Backend），如果後續狀態變得過大的話，就切換到 RocksDB State Backend 中。

在實現層，Flink 現在統一了所有 State Backend 的 Savepoint 格式來實現這一功能。

K8s 部署時使用使用者指定的 Pod 模式

原生 kubernetes 部署（Flink 主動要求 K8s 來啟動 Pod）中，現在可以使用自定義的 Pod 模板。

使用這些模板，使用者可以使用一種更符合 K8s 的方式來設定 JM 和 TM 的 Pod，這種方式比 Flink K8s 整合內建的配置項更加靈活。

生產可用的 Unaligned Checkpoint

Unaligned Checkpoint 目前已達到了生產可用的狀態，我們鼓勵使用者在存在反壓的情況下試用這一功能。

具體來說，Flink 1.13 中引入的這些功能使 Unaligned Checkpoint 更容易使用：

使用者現在使用 Unaligned Checkpoint 時也可以擴縮容應用。如果使用者需要因為效能原因不能使用 Savepoint而必須使用 Retained checkpoint 時，這一功能會非常方便。對於沒有反壓的應用，啟用 Unaligned Checkpoint 現在代價更小。Unaligned Checkpoint 現在可以通過超時來自動觸發，即一個應用預設會使用 Aligned Checkpoint（不儲存傳輸中的資料），而只在對齊超過一定時間範圍時自動切換到 Unaligned Checkpoint（儲存傳輸中的資料）。

關於如何啟用 Unaligned Checkpoint 可以參考相關文件。

機器學習遷移到單獨的倉庫

為了加速 Flink 機器學習的進展（流批統一的機器學習），現在 Flink 機器學習開啟了新的 flink-ml 倉庫。我們採用類似於 Stateful Function 項目的管理方式，通過使用一個單獨的倉庫從而簡化程式碼合併的流程並且可以進行單獨的版本釋出，從而提高開發的效率。

使用者可以關注 Flink 在機器學習方面的進展，比如與 Alink（Flink 常用機器學習演算法套件）的互操作以及 Flink 與 Tensorflow 的整合。

SQL / Table API 進展

與之前的版本類似，SQL 和 Table API 仍然在所有開發中佔用很大的比例。

通過 Table-valued 函數來定義時間視窗

在流式 SQL 查詢中，一個最經常使用的是定義時間視窗。Flink 1.13 中引入了一種新的定義視窗的方式：通過 Table-valued 函數。這一方式不僅有更強的表達能力（允許使用者定義新的視窗類型），並且與 SQL 標準更加一致。

Flink 1.13 在新的語法中支援 TUMBLE 和 HOP 視窗，在後續版本中也會支援 SESSION 視窗。我們通過以下兩個例子來展示這一方法的表達能力：

例 1：一個新引入的 CUMULATE 視窗函數，它可以支援按特定步長擴展的視窗，直到達到最大視窗大小：

SELECT window_time, window_start, window_end, SUM(price) AS total_price FROM TABLE(CUMULATE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '2' MINUTES, INTERVAL '10' MINUTES))GROUP BY window_start, window_end, window_time;

例 2：使用者在 table-valued 視窗函數中可以訪問視窗的起始和終止時間，從而使使用者可以實現新的功能。例如，除了常規的基於視窗的聚合和 Join 之外，使用者現在也可以實現基於視窗的 Top-K 聚合：

SELECT window_time, ... FROM ( SELECT *, ROW_NUMBER() OVER (PARTITION BY window_start, window_end ORDER BY total_price DESC) as rank FROM t ) WHERE rank <= 100;

提高 DataStream API 與 Table API / SQL 的互操作能力

這一版本極大的簡化了 DataStream API 與 Table API 混合的程式。

Table API 是一種非常方便的應用開發介面，因為這經支援表示式的程式編寫並提供了大量的內建函數。但是有時候使用者也需要切換回 DataStream，例如當用戶存在表達能力、靈活性或者 State 訪問的需求時。

Flink 新引入的 StreamTableEnvironment.toDataStream()/.fromDataStream() 可以將一個 DataStream API 聲明的 Source 或者 Sink 當作 Table 的 Source 或者 Sink 來使用。主要的優化包括：

DataStream 與 Table API 類型系統的自動轉換。Event Time 配置的無縫整合，Watermark 行為的高度一致性。Row 類型（即 Table API 中資料的表示）有了極大的增強，包括 toString() / hashCode() 和 equals() 方法的優化，按名稱訪問欄位值的支援與稀疏表示的支援。

Table table = tableEnv.fromDataStream( dataStream, Schema.newBuilder() .columnByMetadata("rowtime", "TIMESTAMP(3)") .watermark("rowtime", "SOURCE_WATERMARK()") .build());DataStream<Row> dataStream = tableEnv.toDataStream(table) .keyBy(r -> r.getField("user")) .window(...);

SQL Client: 初始化指令碼和語句集合 （Statement Sets）

SQL Client 是一種直接運行和部署 SQL 流或批作業的簡便方式，使用者不需要編寫程式碼就可以從命令列呼叫 SQL，或者作為 CI / CD 流程的一部分。

這個版本極大的提高了 SQL Client 的功能。現在基於所有通過 Java 程式設計（即通過程式設計的方式呼叫 TableEnvironment 來發起查詢）可以支援的語法，現在 SQL Client 和 SQL 指令碼都可以支援。這意味著 SQL 使用者不再需要新增膠水程式碼來部署他們的SQL作業。

配置簡化和程式碼共享

Flink 後續將不再支援通過 Yaml 的方式來配置 SQL Client（注：目前還在支援，但是已經被標記為廢棄）。作為替代，SQL Client 現在支援使用一個初始化指令碼在主 SQL 指令碼執行前來配置環境。

這些初始化指令碼通常可以在不同團隊/部署之間共享。它可以用來載入常用的 catalog，應用通用的配置或者定義標準的檢視。

./sql-client.sh -i init1.sql init2.sql -f sqljob.sql

更多的配置項

通過增加配置項，優化 SET / RESET 命令，使用者可以更方便的在 SQL Client 和 SQL 指令碼內部來控制執行的流程。

通過語句集合來支援多查詢

多查詢允許使用者在一個 Flink 作業中執行多個 SQL 查詢（或者語句）。這對於長期運行的流式 SQL 查詢非常有用。

語句集可以用來將一組查詢合併為一組同時執行。

以下是一個可以通過 SQL Client 來執行的 SQL 指令碼的例子。它初始化和配置了執行多查詢的環境。這一指令碼包括了所有的查詢和所有的環境初始化和配置的工作，從而使它可以作為一個自包含的部署元件。

-- set up a catalogCREATE CATALOG hive_catalog WITH ('type' = 'hive');USE CATALOG hive_catalog;-- or use temporary objectsCREATE TEMPORARY TABLE clicks ( user_id BIGINT, page_id BIGINT, viewtime TIMESTAMP) WITH ( 'connector' = 'kafka', 'topic' = 'clicks', 'properties.bootstrap.servers' = '...', 'format' = 'avro');-- set the execution mode for jobsSET execution.runtime-mode=streaming;-- set the sync/async mode for INSERT INTOsSET table.dml-sync=false;-- set the job's parallelismSET parallism.default=10;-- set the job nameSET pipeline.name = my_flink_job;-- restore state from the specific savepoint pathSET execution.savepoint.path=/tmp/flink-savepoints/savepoint-bb0dab;BEGIN STATEMENT SET;INSERT INTO pageview_pv_sinkSELECT page_id, count(1) FROM clicks GROUP BY page_id;INSERT INTO pageview_uv_sinkSELECT page_id, count(distinct user_id) FROM clicks GROUP BY page_id;END;

Hive 查詢語法相容性

使用者現在在 Flink 上也可以使用 Hive SQL 語法。除了 Hive DDL 方言之外，Flink現在也支援常用的 Hive DML 和 DQL 方言。

為了使用 Hive SQL 方言，需要設定 table.sql-dialect 為 hive 並且載入 HiveModule。後者非常重要，因為必須要載入 Hive 的內建函數後才能正確實現對 Hive 語法和語義的相容性。例子如下：

CREATE CATALOG myhive WITH ('type' = 'hive'); -- setup HiveCatalogUSE CATALOG myhive;LOAD MODULE hive; -- setup HiveModuleUSE MODULES hive,core;SET table.sql-dialect = hive; -- enable Hive dialectSELECT key, value FROM src CLUSTER BY key; -- run some Hive queries

需要注意的是， Hive 方言中不再支援 Flink 語法的 DML 和 DQL 語句。如果要使用 Flink 語法，需要切換回 default 的方言配置。

優化的 SQL 時間函數

在資料處理中時間處理是一個重要的任務。但是與此同時，處理不同的時區、日期和時間是一個日益複雜的任務。

在 Flink 1.13 中，我們投入了大量的精力來簡化時間函數的使用。我們調整了時間相關函數的返回類型使其更加精確，例如 PROCTIME()，CURRENT_TIMESTAMP() 和 NOW()。

其次，使用者現在還可以基於一個 TIMESTAMP_LTZ 類型的列來定義 Event Time 屬性，從而可以優雅的在視窗處理中支援夏令時。

使用者可以參考 Release Note 來檢視該部分的完整變更。

PyFlink 核心優化

這個版本對 PyFlink 的改進主要是使基於 Python 的 DataStream API 與 Table API 與 Java/scala 版本的對應功能更加一致。

Python DataStream API 中的有狀態運算元

在 Flink 1.13 中，Python 程式設計師可以享受到 Flink 狀態處理 API 的所有能力。在 Flink 1.12 版本重構過的 Python DataStream API 現在已經擁有完整的狀態訪問能力，從而使使用者可以將資料的資訊記錄到 state 中並且在後續訪問。

帶狀態的處理能力是許多依賴跨記錄狀態共享（例如 Window Operator）的複雜資料處理場景的基礎。

以下例子展示了一個自定義的計算視窗的實現：

class CountWindowAverage(FlatMapFunction): def __init__(self, window_size): self.window_size = window_size def open(self, runtime_context: RuntimeContext): descriptor = ValueStateDescriptor("average", Types.TUPLE([Types.LONG(), Types.LONG()])) self.sum = runtime_context.get_state(descriptor) def flat_map(self, value): current_sum = self.sum.value() if current_sum is None: current_sum = (0, 0) # update the count current_sum = (current_sum[0] + 1, current_sum[1] + value[1]) # if the count reaches window_size, emit the average and clear the state if current_sum[0] >= self.window_size: self.sum.clear() yield value[0], current_sum[1] // current_sum[0] else: self.sum.update(current_sum)ds = ... # type: DataStreamds.key_by(lambda row: row[0]) .flat_map(CountWindowAverage(5))

PyFlink DataStream API 中的使用者自定義視窗

Flink 1.13 中 PyFlink DataStream 介面增加了對使用者自定義視窗的支援，現在使用者可以使用標準視窗之外的視窗定義。

由於視窗是處理無限資料流的核心機制 （通過將流切分為多個有限的『桶』），這一功能極大的提高的 API 的表達能力。

PyFlink Table API 中基於行的操作

Python Table API 現在支援基於行的操作，例如使用者對行資料的自定義函數。這一功能使得使用者可以使用非內建的資料處理函數。

一個使用 map() 操作的 Python Table API 示例如下：

@udf(result_type=DataTypes.ROW( [DataTypes.FIELD("c1", DataTypes.BIGINT()), DataTypes.FIELD("c2", DataTypes.STRING())]))def increment_column(r: Row) -> Row: return Row(r[0] + 1, r[1])table = ... # type: Tablemapped_result = table.map(increment_column)

除了 map()，這一 API 還支援 flat_map()，aggregate()，flat_aggregate() 和其它基於行的操作。這使 Python Table API 的功能與 Java Table API 的功能更加接近。

PyFlink DataStream API 支援 Batch 執行模式

對於有限流，PyFlink DataStream API 現在已經支援 Flink 1.12 DataStream API 中引入的 Batch 執行模式。

通過複用資料有限性來跳過 State backend 和 Checkpoint 的處理，Batch 執行模式可以簡化運維，並且提高有限流處理的效能。

其它優化

基於 Hugo 的 Flink 文件

Flink 文件從 JekyII 遷移到了 Hugo。如果您發現有問題，請務必通知我們，我們非常期待使用者對新的介面的感受。

Web UI 支援歷史異常

Flink Web UI 現在可以展示導致作業失敗的 n 次歷史異常，從而提升在一個異常導致多個後續異常的場景下的偵錯體驗。使用者可以在異常歷史中找到根異常。

優化失敗 Checkpoint 的異常和失敗原因的彙報

Flink 現在提供了失敗或被取消的 Checkpoint 的統計，從而使使用者可以更簡單的判斷 Checkpoint 失敗的原因，而不需要去檢視日誌。

Flink 之前的版本只有在 Checkpoint 成功的時候才會彙報指標（例如持久化資料的大小、觸發時間等）。

提供『恰好一次』一致性的 JDBC Sink

從 1.13 開始，通過使用事務提交資料，JDBC Sink 可以對支援 XA 事務的資料庫提供『恰好一次』的一致性支援。這一特性要求目標資料庫必須有（或連結到）一個 XA 事務處理器。

這一 Sink 現在只能在 DataStream API 中使用。使用者可以通過 JdbcSink.exactlyOnceSink(…) 來創建這一 Sink（或者通過顯式初始化一個 JdbcXaSinkFunction）。

PyFlink Table API 在 Group 視窗上支援使用者自定義的聚合函數

PyFlink Table API 現在對 Group 視窗同時支援基於 Python 的使用者自定義聚合函數（User-defined Aggregate Functions, UDAFs）以及 Pandas UDAFs。這些函數對許多資料分析或機器學習訓練的程式非常重要。

在 Flink 1.13 之前，這些函數僅能在無限的 Group-by 聚合場景下使用。Flink 1.13 優化了這一限制。

Batch 執行模式下 Sort-merge Shuffle 優化

Flink 1.13 優化了針對批處理程式的 Sort-merge Blocking Shuffle 的效能和記憶體佔用情況。這一 Shuffle 模式是在Flink 1.12 的 FLIP-148 中引入的。

這一優化避免了大規模作業下不斷出現 OutOfMemoryError: Direct Memory 的問題，並且通過 I/O 排程和 broadcast 優化提高了效能（尤其是在機械硬碟上）。

HBase 連線器支援非同步維表查詢和查詢快取

HBase Lookup Table Source 現在可以支援非同步查詢模式和查詢快取。這極大的提高了使用這一 Source 的 Table / SQL 維表 Join 的效能，並且在一些典型情況下可以減少對 HBase 的 I/O 請求數量。

在之前的版本中，HBase Lookup Source 僅支援同步通訊，從而導致作業吞吐以及資源利用率降低。

升級 Flink 1.13 需要注意的改動

FLINK-21709 – 老的 Table & SQL API 計劃器已經被標記為廢棄，並且將在 Flink 1.14 中被刪除。Blink 計劃器在若干版本之前已經被設定為預設計劃器，並且將成為未來版本中的唯一計劃器。這意味著 BatchTableEnvironment 和 DataSet API 互操作後續也將不再支援。使用者需要切換到統一的 TableEnvironment 來編寫流或者批的作業。FLINK-22352 – Flink 社群決定廢棄對 Apache mesos 的支援，未來有可能會進一步刪除這部分功能。使用者最好能夠切換到其它的資源管理系統上。FLINK-21935 – state.backend.async 這一配置已經被禁用了，因為現在 Flink 總是會非同步的來儲存快照（即之前的配置預設值），並且現在沒有實現可以支援同步的快照儲存操作。FLINK-17012 – Task 的 RUNNING 狀態被細分為兩步：INITIALIZING 和 RUNNING。Task 的 INITIALIZING 階段包括載入 state 和在啟用 unaligned checkpoint 時恢復 In-flight 資料的過程。通過顯式區分這兩種狀態，監控系統可以更好的區分任務是否已經在實際工作。FLINK-21698 – NUMERIC 和 TIMESTAMP 類型之間的直接轉換存在問題，現在已經被禁用，例如 CAST(numeric AS TIMESTAMP(3))。使用者應該使用 TO_TIMESTAMP(FROM_UNIXTIME(numeric)) 來代替。FLINK-22133 – 新的 Source 介面有一個小的不相容的修改，即 SplitEnumerator.snapshotState() 方法現在多接受一個 checkpoint id 參數來表示正在進行的 snapshot 操作所屬的 checkpoint 的 id。FLINK-19463 – 由於老的 Statebackend 介面承載了過多的語義並且容易引起困惑，這一介面被標記為廢棄。這是一個純 API 層的改動，而並不會影響應用運行時。對於如何升級現有作業，請參考作業遷移指引 。

翻譯 | 高贇Review | 朱翥、馬國維

本文為阿里雲原創內容，未經允許不得轉載。