Flink 1.11 features 已經凍結，流批一體在新版中是濃墨重彩的一筆，在此提前對 Flink 1.11 中流批一體方面的改善進行深度解讀，大家可期待正式版本的發布。
Flink 1.11 中流計算結合 Hive 批處理數倉，給離線數倉帶來 Flink 流處理實時且 Exactly-once 的能力。另外，Flink 1.11 完善了 Flink 自身的 Filesystem connector，大大提高了 Flink 的易用性。
數倉架構
離線數倉
傳統的離線數倉是由 Hive 加上 HDFS 的方案，Hive 數倉有着成熟和穩定的大數據分析能力，結合調度和上下游工具，構建一個完整的數據處理分析平台，流程如下：
- Flume 把數據導入 Hive 數倉
- 調度工具，調度 ETL 作業進行數據處理
- 在 Hive 數倉的表上，可以進行靈活的 Ad-hoc 查詢
- 調度工具，調度聚合作業輸出到 BI 層的數據庫中
這個流程下的問題是：
- 導入過程不夠靈活，這應該是一個靈活 SQL 流計算的過程
- 基於調度作業的級聯計算，實時性太差
- ETL 不能有流式的增量計算
實時數倉
針對離線數倉的特點，隨着實時計算的流行，越來越多的公司引入實時數倉，實時數倉基於 Kafka + Flink streaming，定義全流程的流計算作業，有着秒級甚至毫秒的實時性。
但是，實時數倉的一個問題是歷史數據只有 3-15 天，無法在其上做 Ad-hoc 的查詢。如果搭建 Lambda 的離線 + 實時的架構，維護成本、計算存儲成本、一致性保證、重復的開發會帶來很大的負擔。
Hive 實時化
Flink 1.11 為解決離線數倉的問題，給 Hive 數倉帶來了實時化的能力，加強各環節的實時性的同時，又不會給架構造成太大的負擔。
Hive streaming sink
實時數據導入 Hive 數倉，你是怎么做的？Flume、Spark Streaming 還是 Flink Datastream？千呼萬喚，Table / SQL 層的 streaming file sink 來啦，Flink 1.11 支持 Filesystem connector [1] 和 Hive connector 的 streaming sink [2]。
注：圖中 StreamingFileSink 的 Bucket 概念就是 Table/SQL 中的 Partition
Table/SQL 層的 streaming sink 不僅：
- 帶來 Flink streaming 的實時 / 准實時的能力
- 支持 Filesystem connector 的全部 formats(csv,json,avro,parquet,orc)
- 支持 Hive table 的所有 formats
- 繼承 Datastream StreamingFileSink 的所有特性：Exactly-once、支持 HDFS, S3
而且引入了新的機制：Partition commit。
一個合理的數倉的數據導入，它不止包含數據文件的寫入，也包含了 Partition 的可見性提交。當某個 Partition 完成寫入時，需要通知 Hive metastore 或者在文件夾內添加 SUCCESS 文件。Flink 1.11 的 Partition commit 機制可以讓你：
- Trigger：控制 Partition 提交的時機，可以根據 Watermark 加上從 Partition 中提取的時間來判斷，也可以通過 Processing time 來判斷。你可以控制：是想先盡快看到沒寫完的 Partition；還是保證寫完 Partition 之后，再讓下游看到它。
- Policy：提交策略，內置支持 SUCCESS 文件和 Metastore 的提交，你也可以擴展提交的實現，比如在提交階段觸發 Hive 的 analysis 來生成統計信息，或者進行小文件的合並等等。
一個例子：
-- 結合 Hive dialect 使用 Hive DDL 語法
SET table.sql-dialect=hive;
CREATE TABLE hive_table (
user_id STRING,
order_amount DOUBLE
) PARTITIONED BY (
dt STRING,
hour STRING
) STORED AS PARQUET TBLPROPERTIES (
-- 使用 partition 中抽取時間，加上 watermark 決定 partiton commit 的時機
'sink.partition-commit.trigger'='partition-time',
-- 配置 hour 級別的 partition 時間抽取策略，這個例子中 dt 字段是 yyyy-MM-dd 格式的天，hour 是 0-23 的小時，timestamp-pattern 定義了如何從這兩個 partition 字段推出完整的 timestamp
'partition.time-extractor.timestamp-pattern'=’$dt $hour:00:00’,
-- 配置 dalay 為小時級，當 watermark > partition 時間 + 1 小時，會 commit 這個 partition
'sink.partition-commit.delay'='1 h',
-- partitiion commit 的策略是：先更新 metastore(addPartition)，再寫 SUCCESS 文件
'sink.partition-commit.policy.kind’='metastore,success-file'
)
SET table.sql-dialect=default;
CREATE TABLE kafka_table (
user_id STRING,
order_amount DOUBLE,
log_ts TIMESTAMP(3),
WATERMARK FOR log_ts AS log_ts - INTERVAL '5' SECOND
)
-- 可以結合 Table Hints 動態指定 table properties [3]
INSERT INTO TABLE hive_table SELECT user_id, order_amount, DATE_FORMAT(log_ts, 'yyyy-MM-dd'), DATE_FORMAT(log_ts, 'HH') FROM kafka_table;
Hive streaming source
Hive 數倉中存在大量的 ETL 任務，這些任務往往是通過調度工具來周期性的運行，這樣做主要有兩個問題：
- 實時性不強，往往調度最小是小時級。
- 流程復雜，組件多，容易出現問題。
針對這些離線的 ETL 作業，Flink 1.11 為此開發了實時化的 Hive 流讀，支持：
- Partition 表，監控 Partition 的生成，增量讀取新的 Partition。
- 非 Partition 表，監控文件夾內新文件的生成，增量讀取新的文件。
你甚至可以使用 10 分鍾級別的分區策略，使用 Flink 的 Hive streaming source 和 Hive streaming sink 可以大大提高 Hive 數倉的實時性到准實時分鍾級 [4][5]，在實時化的同時，也支持針對 Table 全量的 Ad-hoc 查詢，提高靈活性。
實時數據關聯 Hive 表
在 Flink 與 Hive 集成的功能發布以后，我們收到最多的用戶反饋之一就是希望能夠將 Flink 的實時數據與離線的 Hive 表進行關聯。因此，在 Flink 1.11 中，我們支持將實時表與 Hive 表進行 temporal join [6]。沿用 Flink 官方文檔中的例子，假定 Orders 是實時表，而 LatestRates 是一張 Hive 表，用戶可以通過以下語句進行 temporal join：
與 Hive 表進行 temporal join 目前只支持 processing time，我們會把 Hive 表的數據緩存到內存中，並按照固定的時間間隔去更新緩存的數據。用戶可以通過參數“lookup.join.cache.ttl” 來控制緩存更新的間隔，默認間隔為一個小時。
“lookup.join.cache.ttl” 需要配置到 Hive 表的 property 當中，因此每張表可以有不同的配置。另外，由於需要將整張 Hive 表加載到內存中，因此目前只適用於 Hive 表較小的場景。
Hive 增強
Hive Dialect 語法兼容
Flink on Hive 用戶並不能很好的使用 DDL，主要是因為：
- Flink 1.10 中進一步完善了 DDL，但由於 Flink 與 Hive 在元數據語義上的差異，通過 Flink DDL 來操作 Hive 元數據的可用性比較差，僅能覆蓋很少的應用場景。
- 使用 Flink 對接 Hive 的用戶經常需要切換到 Hive CLI 來執行 DDL。
針對上述兩個問題，我們提出了 FLIP-123 [7]，通過 Hive Dialect 為用戶提供 Hive 語法兼容。該功能的最終目標，是為用戶提供近似 Hive CLI/Beeline 的使用體驗，讓用戶無需在 Flink 和 Hive 的 CLI 之間進行切換，甚至可以直接遷移部分 Hive 腳本到 Flink 中執行。
在 Flink 1.11 中，Hive Dialect 可以支持大部分常用的 DDL，比如 CREATE/ALTER TABLE、CHANGE/REPLACE COLUMN、ADD/DROP PARTITION 等等。為此，我們為 Hive Dialect 實現了一個獨立的 parser，Flink 會根據用戶指定的 Dialect 決定使用哪個 parser 來解析 SQL 語句。用戶可以通過配置項“ table.sql-dialect ” 來指定使用的 SQL Dialect。它的默認值為 “default”，即 Flink 原生的 Dialect，而將其設置為 “hive” 時就開啟了 Hive Dialect。對於 SQL 用戶，可以在 yaml 文件中設置“table.sql-dialect” 來指定 session 的初始 Dialect，也可以通過 set 命令來動態調整需要使用的 Dialect，而無需重啟 session。
Hive Dialect 目前所支持的具體功能可以參考 FLIP-123 或 Flink 的官方文檔。另外，該功能的一些設計原則和使用注意事項如下：
- Hive Dialect 只能用於操作 Hive 表，而不是 Flink 原生的表（如 Kafka、ES 的表），這也意味着 Hive Dialect 需要配合 HiveCatalog 使用。
- 使用 Hive Dialect 時，原有的 Flink 的一些語法可能會無法使用（例如 Flink 定義的類型別名），在需要使用 Flink 語法時可以動態切換到默認的 Dialect。
- Hive Dialect 的 DDL 語法定義基於 Hive 的官方文檔，而不同 Hive 版本之間語法可能會有輕微的差異，需要二手游戲賬號拍賣地圖用戶進行一定的調整。
- Hive Dialect 的語法實現基於 Calcite，而 Calcite 與 Hive 有不同的保留關鍵字。因此，某些在 Hive 中可以直接作為標識符的關鍵字（如 “default” ），在 Hive Dialect 中可能需要用“`”進行轉義。
向量化讀取
Flink 1.10 中，Flink 已經支持了 ORC (Hive 2+) 的向量化讀取支持，但是這很局限，為此，Flink 1.11 增加了更多的向量化支持：
- ORC for Hive 1.x [8]
- Parquet for Hive 1,2,3 [9]
也就是說已經補全了所有版本的 Parquet 和 ORC 向量化支持，默認是開啟的，提供開關。
簡化 Hive 依賴
Flink 1.10 中，Flink 文檔中列出了所需的 Hive 相關依賴，推薦用戶自行下載。但是這仍然稍顯麻煩，所以在 1.11 中，Flink 提供了內置的依賴支持 [10]：
- flink-sql-connector-hive-1.2.2_2.11-1.11.jar：Hive 1 的依賴版本。
- flink-sql-connector-hive-2.2.0_2.11-1.11.jar：Hive 2.0 - 2.2 的依賴版本。
- flink-sql-connector-hive-2.3.6_2.11-1.11.jar：Hive 2.3 的依賴版本。
- flink-sql-connector-hive-3.1.2_2.11-1.11.jar：Hive 3 的依賴版本。
現在，你只需要單獨下一個包，再搞定 HADOOP_CLASSPATH，即可運行 Flink on Hive。
Flink 增強
除了 Hive 相關的 features，Flink 1.11 也完成了大量其它關於流批一體的增強。
Flink Filesystem connector
Flink table 在長久以來只支持一個 csv 的 file system table，而且它還不支持 Partition，行為上在某些方面也有些不符合大數據計算的直覺。
在 Flink 1.11，重構了整個 Filesystem connector 的實現 [1]：
- 結合 Partition，現在，Filesystem connector 支持 SQL 中 Partition 的所有語義，支持 Partition 的 DDL，支持 Partition Pruning，支持靜態 / 動態 Partition 的插入，支持 overwrite 的插入。
- 支持各種 Formats：CSVJSONAparch AVROApache ParquetApache ORC.
- 支持 Batch 的讀寫。
- 支持 Streaming sink，也支持上述 Hive 支持的 Partition commit，支持寫 Success 文件。
例子：
CREATE TABLE fs_table (
user_id STRING,
order_amount DOUBLE,
dt STRING,
hour STRING
) PARTITIONED BY (dt, hour) WITH (
’connector’=’filesystem’,
’path’=’...’,
’format’=’parquet’,
'partition.time-extractor.timestamp-pattern'=’$dt $hour:00:00’,
'sink.partition-commit.delay'='1 h',
‘sink.partition-commit.policy.kind’='success-file')
)
-- stream environment or batch environment
INSERT INTO TABLE fs_table SELECT user_id, order_amount, DATE_FORMAT(log_ts, 'yyyy-MM-dd'), DATE_FORMAT(log_ts, 'HH') FROM kafka_table;
-- 通過 Partition 查詢
SELECT * FROM fs_table WHERE dt=’2020-05-20’ and hour=’12’;
引入 Max Slot
Yarn perJob 或者 session 模式在 1.11 之前是無限擴張的，沒有辦法限制它的資源使用，只能用 Yarn queue 等方式來限制。但是傳統的批作業其實都是大並發，運行在局限的資源上，一部分一部分階段性的運行，為此，Flink 1.11 引入 Max Slot 的配置 [11]，限制 Yarn application 的資源使用。
slotmanager.number-of-slots.max
定義 Flink 集群分配的最大 Slot 數。此配置選項用於限制批處理工作負載的資源消耗。不建議為流作業配置此選項，如果沒有足夠的 Slot，則流作業可能會失敗。
結語
Flink 1.11 也是一個大版本，社區做了大量的 Features 和 Improvements，Flink 的大目標是幫助業務構建流批一體的數倉，提供完善、順滑、高性能的一體式數倉。希望大家多多參與社區，積極反饋問題和想法，甚至參與社區的討論和開發，一起把 Flink 做得越來越好！