[[440014]]

 1.序篇

廢話不多說，咱們先直接上本文的目錄和結(jié)論，小伙伴可以先看結(jié)論快速了解博主期望本文能給小伙伴們帶來什么幫助：

• 背景及應用場景介紹：博主期望你能了解到，其實很多場景下實時數(shù)倉的建設(shè)都是隨著離線數(shù)倉而建設(shè)的(相同的邏輯在實時數(shù)倉中重新實現(xiàn)一遍)，因此能夠在 flink sql 中復用 hive udf 是能夠大大提高人效的。
• flink 擴展支持 hive 內(nèi)置 udf：flink sql 提供了擴展 udf 的能力，即 module，并且 flink sql 也內(nèi)置了 HiveModule(需要你主動加載進環(huán)境)，來支持一些 hive 內(nèi)置的 udf (比如 get_json_object)給小伙伴們使用。
• flink 擴展支持用戶自定義的 hive udf：主要介紹 flink sql 流任務中，不能使用 create temporary function 去引入一個用戶自定義的 hive udf。因此博主只能通過 flink sql 提供的 module 插件能力，自定義了 module，來支持引入用戶自定義的 hive udf。

2.背景及應用場景介紹

其實大多數(shù)公司都是從離線數(shù)倉開始建設(shè)的。相信大家必然在自己的生產(chǎn)環(huán)境中開發(fā)了非常多的 hive udf。隨著需求對于時效性要求的增高，越來越多的公司也開始建設(shè)起實時數(shù)倉。很多場景下實時數(shù)倉的建設(shè)都是隨著離線數(shù)倉而建設(shè)的。實時數(shù)據(jù)使用 flink 產(chǎn)出，離線數(shù)據(jù)使用 hive\spark 產(chǎn)出。

那么回到我們文章標題的問題：為什么需要 flink 支持 hive udf 呢?

博主分析了下，結(jié)論如下：

站在數(shù)據(jù)需求的角度來說，一般會有以下兩種情況：

• 以前已經(jīng)有了離線數(shù)據(jù)鏈路，需求方也想要實時數(shù)據(jù)。如果直接能用已經(jīng)開發(fā)好的 hive udf，則不用將相同的邏輯遷移到 flink udf 中，并且后續(xù)無需費時費力維護兩個 udf 的邏輯一致性。
• 實時和離線的需求都是新的，需要新開發(fā)。如果只開發(fā)一套 udf，則事半功倍。

因此在 flink 中支持 hive udf 這件事對開發(fā)人員提效來說是非常有好處的。

3.在擴展前，你需要知道一些基本概念

• flink 支持 hive udf 這件事分為兩個部分。
• flink 擴展支持 hive 內(nèi)置 udf

flink 擴展支持用戶自定義 hive udf

第一部分：flink 擴展支持 hive 內(nèi)置 udf，比如 get_json_object，rlike 等等。

有同學問了，這么基本的 udf，flink 都沒有嗎?

確實沒有。關(guān)于 flink sql 內(nèi)置的 udf 見如下鏈接，大家可以看看 flink 支持了哪些 udf：https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/table/functions/systemfunctions/

那么如果我如果強行使用 get_json_object 這個 udf，會發(fā)生啥呢?結(jié)果如下圖。

直接報錯找不到 udf。

第二部分：flink 擴展支持用戶自定義 hive udf。

內(nèi)置函數(shù)解決不了用戶的復雜需求，用戶就需要自己寫 hive udf，并且這部分自定義 udf 也想在 flink sql 中使用。

下面看看怎么在 flink sql 中進行這兩種擴展。

4.hive udf 擴展支持

4.1.flink sql module

涉及到擴展 udf 就不得不提到 flink 提供的 module。見官網(wǎng)下圖。

從第一句話就可以看到，module 的作用就是讓用戶去擴展 udf 的。

flink 本身已經(jīng)內(nèi)置了一個 module，名字叫 CoreModule，其中已經(jīng)包含了一些 udf。

那我們要怎么使用 module 這玩意去擴展我們的 hive udf 呢?

4.2.flink 擴展支持 hive 內(nèi)置 udf

步驟如下：

引入 hive 的 connector。其中包含了 flink 官方提供的一個 HiveModule。在 HiveModule 中包含了 hive 內(nèi)置的 udf。

<dependency> 
    <groupId>org.apache.flink</groupId> 
    <artifactId>flink-connector-hive_${scala.binary.version}</artifactId> 
    <version>${flink.version}</version> 
</dependency> 

在 StreamTableEnvironment 中加載 HiveModule。

String name = "default"; 
String version = "3.1.2"; 
tEnv.loadModule(name, new HiveModule(version)); 

然后在控制臺打印一下目前有的 module。

String[] modules = tEnv.listModules(); 
Arrays.stream(modules).forEach(System.out::println); 

然后可以看到除了 core module，還有我們剛剛加載進去的 default module。

default 
core 

查看所有 module 的所有 udf。在控制臺打印一下。

String[] functions = tEnv.listFunctions(); 
Arrays.stream(functions).forEach(System.out::println); 

就會將 default 和 core module 中的所有包含的 udf 給列舉出來，當然也就包含了 hive module 中的 get_json_object。

然后我們再去在 flink sql 中使用 get_json_object 這個 udf，就沒有報錯，能正常輸出結(jié)果了。

使用 flink hive connector 自帶的 HiveModule，已經(jīng)能夠解決很大一部分常見 udf 使用的問題了。

4.2.flink 擴展支持用戶自定義 hive udf

原本博主是直接想要使用 flink sql 中的 create temporary function 去執(zhí)行引入自定義 hive udf 的。

舉例如下：

CREATE TEMPORARY FUNCTION test_hive_udf as 'flink.examples.sql._09.udf._02_stream_hive_udf.TestGenericUDF'; 

發(fā)現(xiàn)在執(zhí)行這句 sql 時，是可以執(zhí)行成功，將 udf 注冊進去的。

但是在后續(xù) udf 初始化時就報錯了。具體錯誤如下圖。直接報錯 ClassCastException。

看了下源碼，flink 流環(huán)境下(未連接 hive catalog 時)在創(chuàng)建 udf 時會認為這個 udf 是 flink 生態(tài)體系中的 udf。

所以在初始化我們引入的 TestGenericUDF 時，默認會按照 flink 的 UserDefinedFunction 強轉(zhuǎn)，因此才會報強轉(zhuǎn)錯誤。

那么我們就不能使用 hive udf 了嗎?

錯誤，小伙伴萌豈敢有這種想法。博主都把這個標題列出來了(牛逼都吹出去了)，還能給不出解決方案嘛。

思路見下一章節(jié)。

4.3.flink 擴展支持用戶自定義 hive udf 的增強 module

其實思路很簡單。

使用 flink sql 中的 create temporary function 雖然不能執(zhí)行，但是 flink 提供了插件化的自定義 module。

我們可以擴展一個支持用戶自定義 hive udf 的 module，使用這個 module 來支持自定義的 hive udf。

實現(xiàn)的代碼也非常簡單。簡單的把 flink hive connector 提供的 HiveModule 做一個增強即可，即下圖中的 HiveModuleV2。

使用方式如下圖所示：

然后程序就正常跑起來了。

肥腸滴好用!

5.總結(jié)與展望

本文主要介紹了如果在 flink sql 使用 hive 內(nèi)置 udf 及用戶自定義 hive udf，總結(jié)如下：

• 背景及應用場景介紹：博主期望你能了解到，其實很多場景下實時數(shù)倉的建設(shè)都是隨著離線數(shù)倉而建設(shè)的(相同的邏輯在實時數(shù)倉中重新實現(xiàn)一遍)，因此能夠在 flink sql 中復用 hive udf 是能夠大大提高人效的。
• flink 擴展支持 hive 內(nèi)置 udf：flink sql 提供了擴展 udf 的能力，即 module，并且 flink sql 也內(nèi)置了 HiveModule(需要你主動加載進環(huán)境)，來支持一些 hive 內(nèi)置的 udf (比如 get_json_object)給小伙伴們使用。
• flink 擴展支持用戶自定義的 hive udf：主要介紹 flink sql 流任務中，不能使用 create temporary function 去引入一個用戶自定義的 hive udf。因此博主只能通過 flink sql 提供的 module 插件能力，自定義了 module，來支持引入用戶自定義的 hive udf。