Flink SQL 知其所以然：在 Flink 中還能使用 Hive Udf？（附源碼）
1.序篇
廢話不多說，咱們先直接上本文的目錄和結論，小伙伴可以先看結論快速了解博主期望本文能給小伙伴們帶來什么幫助：
- 背景及應用場景介紹：博主期望你能了解到，其實很多場景下實時數倉的建設都是隨著離線數倉而建設的(相同的邏輯在實時數倉中重新實現一遍)，因此能夠在 flink sql 中復用 hive udf 是能夠大大提高人效的。
- flink 擴展支持 hive 內置 udf：flink sql 提供了擴展 udf 的能力，即 module，并且 flink sql 也內置了 HiveModule(需要你主動加載進環境)，來支持一些 hive 內置的 udf (比如 get_json_object)給小伙伴們使用。
- flink 擴展支持用戶自定義的 hive udf：主要介紹 flink sql 流任務中，不能使用 create temporary function 去引入一個用戶自定義的 hive udf。因此博主只能通過 flink sql 提供的 module 插件能力，自定義了 module，來支持引入用戶自定義的 hive udf。
2.背景及應用場景介紹
其實大多數公司都是從離線數倉開始建設的。相信大家必然在自己的生產環境中開發了非常多的 hive udf。隨著需求對于時效性要求的增高，越來越多的公司也開始建設起實時數倉。很多場景下實時數倉的建設都是隨著離線數倉而建設的。實時數據使用 flink 產出，離線數據使用 hive\spark 產出。
那么回到我們文章標題的問題：為什么需要 flink 支持 hive udf 呢?
博主分析了下，結論如下：
站在數據需求的角度來說，一般會有以下兩種情況：
- 以前已經有了離線數據鏈路，需求方也想要實時數據。如果直接能用已經開發好的 hive udf，則不用將相同的邏輯遷移到 flink udf 中，并且后續無需費時費力維護兩個 udf 的邏輯一致性。
- 實時和離線的需求都是新的，需要新開發。如果只開發一套 udf，則事半功倍。
因此在 flink 中支持 hive udf 這件事對開發人員提效來說是非常有好處的。
3.在擴展前，你需要知道一些基本概念
- flink 支持 hive udf 這件事分為兩個部分。
- flink 擴展支持 hive 內置 udf
flink 擴展支持用戶自定義 hive udf
第一部分：flink 擴展支持 hive 內置 udf，比如 get_json_object，rlike 等等。
有同學問了，這么基本的 udf，flink 都沒有嗎?
確實沒有。關于 flink sql 內置的 udf 見如下鏈接，大家可以看看 flink 支持了哪些 udf：https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/table/functions/systemfunctions/
那么如果我如果強行使用 get_json_object 這個 udf，會發生啥呢?結果如下圖。
直接報錯找不到 udf。
第二部分：flink 擴展支持用戶自定義 hive udf。
內置函數解決不了用戶的復雜需求，用戶就需要自己寫 hive udf，并且這部分自定義 udf 也想在 flink sql 中使用。
下面看看怎么在 flink sql 中進行這兩種擴展。
4.hive udf 擴展支持
4.1.flink sql module
涉及到擴展 udf 就不得不提到 flink 提供的 module。見官網下圖。
從第一句話就可以看到，module 的作用就是讓用戶去擴展 udf 的。
flink 本身已經內置了一個 module，名字叫 CoreModule，其中已經包含了一些 udf。
那我們要怎么使用 module 這玩意去擴展我們的 hive udf 呢?
4.2.flink 擴展支持 hive 內置 udf
步驟如下：
引入 hive 的 connector。其中包含了 flink 官方提供的一個 HiveModule。在 HiveModule 中包含了 hive 內置的 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，就沒有報錯，能正常輸出結果了。
使用 flink hive connector 自帶的 HiveModule，已經能夠解決很大一部分常見 udf 使用的問題了。
4.2.flink 擴展支持用戶自定義 hive udf
原本博主是直接想要使用 flink sql 中的 create temporary function 去執行引入自定義 hive udf 的。
舉例如下：
- CREATE TEMPORARY FUNCTION test_hive_udf as 'flink.examples.sql._09.udf._02_stream_hive_udf.TestGenericUDF';
發現在執行這句 sql 時，是可以執行成功，將 udf 注冊進去的。
但是在后續 udf 初始化時就報錯了。具體錯誤如下圖。直接報錯 ClassCastException。
看了下源碼，flink 流環境下(未連接 hive catalog 時)在創建 udf 時會認為這個 udf 是 flink 生態體系中的 udf。
所以在初始化我們引入的 TestGenericUDF 時，默認會按照 flink 的 UserDefinedFunction 強轉，因此才會報強轉錯誤。
那么我們就不能使用 hive udf 了嗎?
錯誤，小伙伴萌豈敢有這種想法。博主都把這個標題列出來了(牛逼都吹出去了)，還能給不出解決方案嘛。
思路見下一章節。
4.3.flink 擴展支持用戶自定義 hive udf 的增強 module
其實思路很簡單。
使用 flink sql 中的 create temporary function 雖然不能執行，但是 flink 提供了插件化的自定義 module。
我們可以擴展一個支持用戶自定義 hive udf 的 module，使用這個 module 來支持自定義的 hive udf。
實現的代碼也非常簡單。簡單的把 flink hive connector 提供的 HiveModule 做一個增強即可，即下圖中的 HiveModuleV2。
使用方式如下圖所示：
然后程序就正常跑起來了。
肥腸滴好用!
5.總結與展望
本文主要介紹了如果在 flink sql 使用 hive 內置 udf 及用戶自定義 hive udf，總結如下：
- 背景及應用場景介紹：博主期望你能了解到，其實很多場景下實時數倉的建設都是隨著離線數倉而建設的(相同的邏輯在實時數倉中重新實現一遍)，因此能夠在 flink sql 中復用 hive udf 是能夠大大提高人效的。
- flink 擴展支持 hive 內置 udf：flink sql 提供了擴展 udf 的能力，即 module，并且 flink sql 也內置了 HiveModule(需要你主動加載進環境)，來支持一些 hive 內置的 udf (比如 get_json_object)給小伙伴們使用。
- flink 擴展支持用戶自定義的 hive udf：主要介紹 flink sql 流任務中，不能使用 create temporary function 去引入一個用戶自定義的 hive udf。因此博主只能通過 flink sql 提供的 module 插件能力，自定義了 module，來支持引入用戶自定義的 hive udf。
