導(dǎo)讀：隨著 Flink 在流式計(jì)算的應(yīng)用場(chǎng)景逐漸成熟和流行，如果 Flink 能同時(shí)把批量計(jì)算的應(yīng)用場(chǎng)景處理好，就能減少用戶在使用 Flink 時(shí)開(kāi)發(fā)和維護(hù)的成本,并且能夠豐富 Flink 的生態(tài)。SQL 是批計(jì)算中比較常用的工具，所以 Flink 針對(duì)于批計(jì)算也以 SQL 為主要接口。本次分享主要介紹 Flink 對(duì)批處理的設(shè)計(jì)與 Hive 的集成。主要分為下面三點(diǎn)展開(kāi)：

• 設(shè)計(jì)架構(gòu)

• 項(xiàng)目進(jìn)展

• 性能測(cè)試

設(shè)計(jì)架構(gòu)

首先和大家分享一下 Flink 批處理的設(shè)計(jì)架構(gòu)。

1.??背景

Flink 提升批處理的主要原因是為了減少客戶的維護(hù)成本和更新成本和更好的完善 Flink 生態(tài)環(huán)境。SQL 是批計(jì)算場(chǎng)景中一個(gè)非常重要的工具，所以希望以 SQL 作為在批計(jì)算場(chǎng)景的主要接口，為此我們著重優(yōu)化了 Flink SQL 的功能。當(dāng)前 Flink SQL 主要有下面幾點(diǎn)需要優(yōu)化：

• 需要完整的元數(shù)據(jù)管理體制。

• 缺少對(duì) DDL(數(shù)據(jù)定義語(yǔ)言 DDL 用來(lái)創(chuàng)建數(shù)據(jù)庫(kù)中的各種對(duì)象，如表、視圖、索引、同義詞、聚簇等)的支持。

• 與外部系統(tǒng)進(jìn)行對(duì)接不是很方便，尤其是 Hive, 因?yàn)?Hive 是大數(shù)據(jù)領(lǐng)域最早的 SQL 引擎，所以 Hive 的用戶基礎(chǔ)非常廣泛，新的一些 SQL 工具，如 Spark SQL、Impala 都提供了與 Hive 對(duì)接的功能，這樣用戶才能更好地將其應(yīng)用從 Hive 遷移過(guò)來(lái)，所以與 Hive 對(duì)接對(duì) Flink SQL 而言也十分重要。

2.??目標(biāo)

所以我們要完成以下目標(biāo)：

• 定義統(tǒng)一的 Catalog 接口，這個(gè)是 Flink SQL 更方便與外部對(duì)接的前提條件。如果大家用過(guò) Flink 的 TableSource 和 TableSink 來(lái)對(duì)接外部的系統(tǒng)的表，會(huì)發(fā)現(xiàn)不管是通過(guò)寫(xiě)程序還是配置 yaml 文件會(huì)跟傳統(tǒng)的 SQL 使用方式會(huì)有些不同。所以我們肯定不希望 Hive 的用戶遷移 Flink SQL 需要通過(guò)定義 TableSouces 和 TableSink 的方式來(lái)與 Hive 進(jìn)行交互。因此我們提供了一套新的 Catalog 接口以一種更接近傳統(tǒng) SQL 的方式與 Hive 進(jìn)行交互。

• 提供基于內(nèi)存和可持久化的實(shí)現(xiàn)。基于內(nèi)存就是 Flink 原有的方式，用戶所有的元數(shù)據(jù)的生命周期是跟他的 Session（會(huì)話）綁定的，Session（會(huì)話）結(jié)束之后所有的元數(shù)據(jù)都沒(méi)有了。因?yàn)橐?Hive 交互所以肯定還要提供一個(gè)持久化的 Catalog。

• 支持 Hive 的互操作。有了 Catalog 之后用戶就可以通過(guò) Catalog 訪問(wèn) Hive 的元數(shù)據(jù)，提供 Data Connector 讓用戶能通過(guò) Flink 讀寫(xiě) Hive 的實(shí)際數(shù)據(jù)，實(shí)現(xiàn) Flink 與 Hive 的交互。

• 支持 Flink 作為 Hive 的計(jì)算引擎（長(zhǎng)期目標(biāo)），像 Hive On Spark，Hive On Tez。

3.??全新設(shè)計(jì)的 Catalog API（FlIP-30）

用戶通過(guò) SQL Client 或者 Table API 提交請(qǐng)求，Flink 會(huì)創(chuàng)建 TableEnvironment， TableEnvironment 會(huì)創(chuàng)建 CatalogManager 加載并配置 Catalog 實(shí)例，并且 Catalog 支持多種元數(shù)據(jù)類型 table、database、function、view、partition 等，在 1.9.0 的版本當(dāng)中 Catalog 會(huì)有兩個(gè)實(shí)現(xiàn)：

• 一個(gè)是基于內(nèi)存的 GenericinMemoryCatalog。

• 另一是 HiveCatalog，HiveCatalog 通過(guò) HiveShim 與 Hive Metasotre 交互來(lái)操作 Hive 元數(shù)據(jù)，HiveShim 的作用是處理 Hive 在大版本中 Hive Metastore 不兼容的問(wèn)題。

從這種實(shí)現(xiàn)的方式可以看出，用戶可以創(chuàng)建多個(gè) Catalog，也可以訪問(wèn)多個(gè) Hive Metastore，來(lái)達(dá)到跨 Catalog 查詢的操作。

4.??讀寫(xiě) Hive 數(shù)據(jù)

有了元數(shù)據(jù)之后我們就可以實(shí)現(xiàn) Flink SQL 的 Data Connector 來(lái)真正的讀寫(xiě) Hive 實(shí)際數(shù)據(jù)。Flink SQL 寫(xiě)入的數(shù)據(jù)必須要兼容 Hive 的數(shù)據(jù)格式，也就是 Hive 可以正常讀取 Flink 寫(xiě)入的數(shù)據(jù)，反過(guò)來(lái)也是一樣的。為了實(shí)現(xiàn)這一點(diǎn)我們大量復(fù)用 Hive 原有的 Input/Output Format、SerDe 等 API，一是為了減少代碼冗余，二是盡可能的保持兼容性。

在 Data Connect 中讀取 Hive 表數(shù)據(jù)具體實(shí)現(xiàn)類為：HiveTableSource、HiveTableInputFormat。寫(xiě) Hive 表的具體實(shí)現(xiàn)類為：HiveTableSink、HiveTableOutputFormat。

項(xiàng)目進(jìn)展

其次和大家分享 Flink 1.9.0 的現(xiàn)狀和 1.10.0 中的新特性還有未來(lái)工作。

1.?Flink 1.9.0 的現(xiàn)狀

Flink SQL 作為 1.9.0 版本中作為試用功能發(fā)布的，它的功能還不是很完善：

• 支持的數(shù)據(jù)類型還不全。（1.9.0 中帶參數(shù)的數(shù)據(jù)類型基本上都不支持：如 DECIMAL,CHAR 等)

• 對(duì)分區(qū)表的支持不完善，只能讀取分區(qū)表，不能寫(xiě)分區(qū)表。

• 不支持表的 INSERT OVERWRITE。

2.?Flink 1.10.0 中的新特性

Flink SQL 在 1.10.0 版本里我們做了比較多的進(jìn)一步開(kāi)發(fā)，與 Hive 集成的功能更加完整。

• 支持讀寫(xiě)靜態(tài)分區(qū)和動(dòng)態(tài)分區(qū)表。

• 在表級(jí)別和分區(qū)級(jí)別都支持 INSERT OVERWRITE。

• 支持了更多地?cái)?shù)據(jù)類型。（除 UNION 類型都支持）

• 支持更多地 DDL。（CREATE TABLE/DATABASE）

• 支持在 Flink 中調(diào)用 Hive 的內(nèi)置函數(shù)。（Hive 大約 200 多個(gè)內(nèi)置函數(shù)）

• 支持了更多的 Hive 版本。（Hive 的 1.0.0~3.1.1）

• 做了很多性能優(yōu)化如，Project/Predicate Pushdown，向量的讀取 ORC 數(shù)據(jù)等。

3.?Module 接口

為了能讓用戶調(diào)用 Flink SQL 中調(diào)用 Hive 的內(nèi)置函數(shù)，我們?cè)?Flink 1.10 當(dāng)中引入了一個(gè) Module 接口。這個(gè) Module 是為了讓用戶能夠方便的把外部系統(tǒng)的內(nèi)置函數(shù)接入到系統(tǒng)當(dāng)中。

• 使用方式和 Catalog 類似，用戶可以通過(guò) Table API 或 Yaml 文件來(lái)配置 Module。

• Module 可以同時(shí)加載多個(gè)，Flink 解析函數(shù)的時(shí)候通過(guò) Module 的加載順序在多個(gè) Module 中查找函數(shù)的解析。也就是如果兩個(gè) Module 包含名字相同的 Function，先加載的 Module 會(huì)提供 Function 的定義。

• 目前 Module 有兩個(gè)實(shí)現(xiàn)，CoreModule 提供了 Flink 原生的內(nèi)置函數(shù)，HiveModule 提供了 Hive 的內(nèi)置函數(shù)。

4.?未來(lái)工作

未來(lái)的工作主要是先做功能的補(bǔ)全，其中包括：

• View 的支持（有可能在 1.11 中完成）。

• 持續(xù)改進(jìn) SQL CLI 的易用性，現(xiàn)在支持翻頁(yè)顯示查詢結(jié)果，后續(xù)支持滾動(dòng)顯示。并支持 Hive 的 -e -f 這種非交互式的使用方式。

• 支持所有的 Hive 常用 DDL，例如 CREATE TABLE AS。

• 兼容 Hive 的語(yǔ)法，讓原來(lái)在 Hive 上的工程在 Flink 的順滑的遷移過(guò)來(lái)。

• 支持 SQL CLI 的遠(yuǎn)程模式，類似 HiveServer2 的遠(yuǎn)程連接模式。

• 支持流式的寫(xiě)入 Hive 數(shù)據(jù)。

性能測(cè)試

下面是 Flink 在批處理作業(yè)下與 HiveMR 對(duì)比測(cè)試的測(cè)試環(huán)境和結(jié)果。

1.?測(cè)試環(huán)境

首先我們的測(cè)試環(huán)境使用了 21 個(gè)節(jié)點(diǎn)的物理機(jī)群，一個(gè) Master 節(jié)點(diǎn)和 20 個(gè) Slave 節(jié)點(diǎn)。節(jié)點(diǎn)的硬件配置是 32 核，64 個(gè)線程，256 內(nèi)存，網(wǎng)絡(luò)做了端口聚合，每個(gè)機(jī)器是 12 塊的 HDD 硬盤。

2.?測(cè)試工具??

測(cè)試工具使用了 Hortonworks 的 hive-testbench，github 中一個(gè)開(kāi)源的工具。我們使用這個(gè)工具生成了 10TB 的 TPC-DS 測(cè)試數(shù)據(jù)集，然后分別通過(guò) Flink SQL 和 Hive 對(duì)該數(shù)據(jù)集進(jìn)行 TPC-DS 的測(cè)試。

一方面我們對(duì)比了 Flink 和 Hive 的性能，另一方面我們驗(yàn)證了 Flink SQL 能夠很好的訪問(wèn) Hive 的數(shù)據(jù)。測(cè)試用到了 Hive 版本是 3.1.1，Flink 用到的是 Master 分支代碼。

3.?測(cè)試結(jié)果

測(cè)試結(jié)果 Flink SQL 對(duì)比 Hive On MapReduce 取得了大約 7 倍的性能提升。這得益于 Flink SQL 所做的一系列優(yōu)化，比如在調(diào)度方面的優(yōu)化，以及執(zhí)行計(jì)劃的優(yōu)化等。總體來(lái)說(shuō)如果用的是 Hive On MapReduce，遷移到 Flink SQL 會(huì)有很大性能的提升。

總結(jié)

以上是生活随笔為你收集整理的性能提升约 7 倍！Apache Flink 与 Apache Hive 的集成的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。