UDF 是什么？

Hive 中的 UDF 其實(shí)就是用戶自定義函數(shù)，允許用戶注冊使用自定義的邏輯對數(shù)據(jù)進(jìn)行處理，豐富了Hive 對數(shù)據(jù)處理的能力。

UDF 負(fù)責(zé)完成對數(shù)據(jù)一進(jìn)一出處理的操作，和 Hive 中存在的函數(shù) year、month、day 等相同。

reflect

在 Hive 中，可以使用 reflect() 方法通過 Java 反射機(jī)制調(diào)用 Java 類的方法。

通俗來說，它可以調(diào)用 Hive 中不存在，但是 JDK 中擁有的方法。

語法

• reflect() 函數(shù)的語法為：reflect(class,method[,arg1[,arg2..]])。

靜態(tài)方法調(diào)用

假設(shè)當(dāng)前在 Java 中存在類如下：

package com.example;public class MathUtils {public static int addNumbers(int a, int b) {return a + b;}
}

那么使用 reflect() 方法調(diào)用時，如下所示：

SELECT reflect("com.example.MathUtils", "addNumbers", 3, 5) AS result;

注意！ 這里的類 "com.example.MathUtils" 并不是在 JDK 中真實(shí)存在的，只是我作為說明的一個案例， reflect() 方法只能調(diào)用 JDK 中（原生內(nèi)置）存在的方法。

所以當(dāng)你需要使用 reflect() 方法時，需要先去查找調(diào)用的目標(biāo)方法全類名、方法名以及是否需要傳遞參數(shù)。

實(shí)例方法調(diào)用

當(dāng)我們需要調(diào)用 Java 中的實(shí)例方法時，先創(chuàng)建 Java 對象，然后再調(diào)用其方法。

例如：將亂碼的字符串進(jìn)行解析。

SELECT reflect('java.net.URLDecoder', 'decode', "Mozilla/5.0%20(compatible;%20MJ12bot/v1.4.7;%20http://www.majestic12.co.uk/bot.php?+)
" ,'utf-8') as result;

結(jié)果輸出如下：

自定義 UDF（GenericUDF）

Hive 支持兩種 UDF 函數(shù)自定義操作，分別是：

• GenericUDF（通用UDF）：用于實(shí)現(xiàn)那些可以處理任意數(shù)據(jù)類型的函數(shù)。它們的輸入和輸出類型可以是任意的，但需要在函數(shù)內(nèi)部處理類型轉(zhuǎn)換和邏輯，可以實(shí)現(xiàn)更復(fù)雜的邏輯處理。

• UDF：用于實(shí)現(xiàn)那些只能處理特定數(shù)據(jù)類型的函數(shù)。每個 UDF 都明確指定了輸入?yún)?shù)的類型和返回值類型，使用更為簡單。

本文采用的是通用 UDF —— GenericUDF 實(shí)現(xiàn)方法

這里通過一個在 Hive 中實(shí)現(xiàn)兩數(shù)相加的自定義 UDF 案例來進(jìn)行說明，看完你就會啦，輕松拿捏。

1.創(chuàng)建項(xiàng)目

在 IDEA 中創(chuàng)建一個 Maven 項(xiàng)目，引入 Hive 依賴，如下所示：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>org.jsu</groupId><artifactId>MyUDF</artifactId><version>1.0-SNAPSHOT</version><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target></properties><dependencies><!-- hive-exec依賴無需打到j(luò)ar包，故scope使用provided--><dependency><groupId>org.apache.hive</groupId><artifactId>hive-exec</artifactId><version>3.1.3</version><scope>provided</scope></dependency></dependencies><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-assembly-plugin</artifactId><version>3.0.0</version><configuration><!--將依賴編譯到j(luò)ar包中--><descriptorRefs><descriptorRef>jar-with-dependencies</descriptorRef></descriptorRefs></configuration><executions><!--配置執(zhí)行器--><execution><id>make-assembly</id><!--綁定到package執(zhí)行周期上--><phase>package</phase><goals><!--只運(yùn)行一次--><goal>single</goal></goals></execution></executions></plugin></plugins></build></project>

注意，引入的 Hive 依賴版本請保持和你集群中使用的版本一致。

2.創(chuàng)建類繼承 UDF

創(chuàng)建一個類，我這里取名為 AddTest，繼承 Hive UDF 父類 GenericUDF，需要重寫三個方法，如下所示：

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;public class AddTest extends GenericUDF {@Overridepublic ObjectInspector initialize(ObjectInspector[] objectInspectors) throws UDFArgumentException {return null;}@Overridepublic Object evaluate(DeferredObject[] deferredObjects) throws HiveException {return null;}@Overridepublic String getDisplayString(String[] strings) {return null;}
}

• initialize(ObjectInspector[] objectInspectors) 方法
  這個方法是在 UDF 初始化時調(diào)用的。它用于執(zhí)行一些初始化操作，并且可以用來驗(yàn)證 UDF 的輸入?yún)?shù)類型是否正確。參數(shù) objectInspectors 是一個包含輸入?yún)?shù)的 ObjectInspector 數(shù)組，它描述了每個輸入?yún)?shù)的類型和結(jié)構(gòu)。
  一般在這個方法中檢查輸入?yún)?shù)的數(shù)量和類型是否滿足你的函數(shù)的要求。如果輸入?yún)?shù)不符合預(yù)期，你可以拋出 UDFArgumentException 異常。如果一切正常，你需要返回一個合適的 ObjectInspector 對象，它描述了你的函數(shù)返回值的類型。

• evaluate(DeferredObject[] deferredObjects) 方法
  在這個方法中定義真正執(zhí)行 UDF 邏輯的地方，獲取輸入的參數(shù)，并且根據(jù)輸入?yún)?shù)執(zhí)行相應(yīng)的計(jì)算或操作。參數(shù) deferredObjects 是一個包含輸入?yún)?shù)的 DeferredObject 數(shù)組，你可以通過它來獲取實(shí)際的輸入值。

• getDisplayString(String[] strings) 方法
  這個方法用于描述 UDF 的信息，用于生成可讀的查詢執(zhí)行計(jì)劃（Explain），以便用戶了解查詢的結(jié)構(gòu)和執(zhí)行過程。

3.數(shù)據(jù)類型判斷

實(shí)現(xiàn) UDF 的第一步操作就是在 initialize 方法中，判斷用戶輸入的參數(shù)是否合法，出現(xiàn)錯誤時，進(jìn)行反饋。

在這里主要分為三個步驟：

1. 檢驗(yàn)參數(shù)個數(shù)

2. 檢查參數(shù)類型

3. 定義函數(shù)返回值類型

一般情況下，可以使用下面的模板：

    @Overridepublic ObjectInspector initialize(ObjectInspector[] objectInspectors) throws UDFArgumentException {// 1.校驗(yàn)參數(shù)個數(shù)if (objectInspectors.length != 2) {throw new UDFArgumentException("參數(shù)個數(shù)有誤！");}// 2.檢查第1個參數(shù)是否是int類型// 判斷第1個參數(shù)的基本類型ObjectInspector num1 = objectInspectors[0];if (num1.getCategory() != ObjectInspector.Category.PRIMITIVE) {throw new UDFArgumentException("第1個參數(shù)不是基本數(shù)據(jù)類型");}// 第1個參數(shù)類型判斷PrimitiveObjectInspector temp = (PrimitiveObjectInspector) num1;if (PrimitiveObjectInspector.PrimitiveCategory.INT != temp.getPrimitiveCategory()) {throw new UDFArgumentException("第1個參數(shù)應(yīng)為INT類型");}// 2.檢查第2個參數(shù)是否是int類型// 判斷第2個參數(shù)的基本類型ObjectInspector num2 = objectInspectors[1];if (num2.getCategory() != ObjectInspector.Category.PRIMITIVE) {throw new UDFArgumentException("第2個參數(shù)不是基本數(shù)據(jù)類型");}// 第2個參數(shù)類型判斷PrimitiveObjectInspector temp2 = (PrimitiveObjectInspector) num2;if (PrimitiveObjectInspector.PrimitiveCategory.INT != temp2.getPrimitiveCategory()) {throw new UDFArgumentException("第2個參數(shù)應(yīng)為INT類型");}// 3.設(shè)置函數(shù)返回值類型（返回一個整型數(shù)據(jù)）return PrimitiveObjectInspectorFactory.javaIntObjectInspector;}

4.編寫業(yè)務(wù)邏輯

在 evaluate 方法中定義業(yè)務(wù)邏輯，這里比較簡單，就是實(shí)現(xiàn)兩數(shù)相加。

    @Overridepublic Object evaluate(DeferredObject[] deferredObjects) throws HiveException {// 完成兩數(shù)相加的邏輯計(jì)算int num1 = Integer.parseInt(deferredObjects[0].get().toString());int num2 = Integer.parseInt(deferredObjects[1].get().toString());return num1 + num2;}

5.定義函數(shù)描述信息

在 getDisplayString 方法中定義函數(shù)在 Explain 中的描述信息，一般都是固定寫法，如下所示：

    @Overridepublic String getDisplayString(String[] strings) {return getStandardDisplayString("AddTest", strings);}

把對應(yīng)的函數(shù)名稱進(jìn)行替換即可。

6.打包與上傳

對編寫的項(xiàng)目進(jìn)行打包，并上傳到 HDFS 上。

本案例的完整代碼如下所示：

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;public class AddTest extends GenericUDF {@Overridepublic ObjectInspector initialize(ObjectInspector[] objectInspectors) throws UDFArgumentException {// 1.校驗(yàn)參數(shù)個數(shù)if (objectInspectors.length != 2) {throw new UDFArgumentException("參數(shù)個數(shù)有誤！");}// 2.檢查第1個參數(shù)是否是int類型// 判斷第1個參數(shù)的基本類型ObjectInspector num1 = objectInspectors[0];if (num1.getCategory() != ObjectInspector.Category.PRIMITIVE) {throw new UDFArgumentException("第1個參數(shù)不是基本數(shù)據(jù)類型");}// 第1個參數(shù)類型判斷PrimitiveObjectInspector temp = (PrimitiveObjectInspector) num1;if (PrimitiveObjectInspector.PrimitiveCategory.INT != temp.getPrimitiveCategory()) {throw new UDFArgumentException("第1個參數(shù)應(yīng)為INT類型");}// 2.檢查第2個參數(shù)是否是int類型// 判斷第2個參數(shù)的基本類型ObjectInspector num2 = objectInspectors[1];if (num2.getCategory() != ObjectInspector.Category.PRIMITIVE) {throw new UDFArgumentException("第2個參數(shù)不是基本數(shù)據(jù)類型");}// 第2個參數(shù)類型判斷PrimitiveObjectInspector temp2 = (PrimitiveObjectInspector) num2;if (PrimitiveObjectInspector.PrimitiveCategory.INT != temp2.getPrimitiveCategory()) {throw new UDFArgumentException("第2個參數(shù)應(yīng)為INT類型");}// 3.設(shè)置函數(shù)返回值類型（返回一個整型數(shù)據(jù)）return PrimitiveObjectInspectorFactory.javaIntObjectInspector;}@Overridepublic Object evaluate(DeferredObject[] deferredObjects) throws HiveException {// 完成兩數(shù)相加的邏輯計(jì)算int num1 = Integer.parseInt(deferredObjects[0].get().toString());int num2 = Integer.parseInt(deferredObjects[1].get().toString());return num1 + num2;}@Overridepublic String getDisplayString(String[] strings) {return getStandardDisplayString("AddTest", strings);}}

7.注冊 UDF 函數(shù)并測試

進(jìn)入 Hive 中對創(chuàng)建的 UDF 函數(shù)進(jìn)行注冊。

如果你期間修改了 JAR 包并重新上傳，則需要重啟與 Hive 的連接，建立新的會話才會生效。

-- 永久注冊
create function testAdd as 'AddTest' using jar 'hdfs://hadoop201:8020/test/MyUDF-1.0-SNAPSHOT-jar-with-dependencies.jar';-- 刪除注冊的函數(shù)
drop function if exists testAdd;

• testAdd：注冊的 UDF 函數(shù)名稱。

• as 'AddTest'：編寫的 UDF 函數(shù)全類名。

• using jar：指定 JAR 包的全路徑。

注冊成功后，如下所示：

測試

select testAdd(1,2);

如果輸入錯誤的數(shù)據(jù)類型，會進(jìn)行報錯提示：

返回復(fù)雜的數(shù)據(jù)類型

在更多的場景下，我們可能有多個返回值，那么該如何定義與配置呢？

這里還是通過上面的兩數(shù)相加的案例來進(jìn)行說明，套下面的模板使用：

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;import java.util.ArrayList;public class AddTestReturnList extends GenericUDF {@Overridepublic ObjectInspector initialize(ObjectInspector[] objectInspectors) throws UDFArgumentException {// 1.校驗(yàn)參數(shù)個數(shù)if (objectInspectors.length != 2) {throw new UDFArgumentException("參數(shù)個數(shù)有誤！");}// 2.檢查第1個參數(shù)是否是int類型// 判斷第1個參數(shù)的基本類型ObjectInspector num1 = objectInspectors[0];if (num1.getCategory() != ObjectInspector.Category.PRIMITIVE) {throw new UDFArgumentException("第1個參數(shù)不是基本數(shù)據(jù)類型");}// 第1個參數(shù)類型判斷PrimitiveObjectInspector temp = (PrimitiveObjectInspector) num1;if (PrimitiveObjectInspector.PrimitiveCategory.INT != temp.getPrimitiveCategory()) {throw new UDFArgumentException("第1個參數(shù)應(yīng)為INT類型");}// 2.檢查第2個參數(shù)是否是int類型// 判斷第2個參數(shù)的基本類型ObjectInspector num2 = objectInspectors[1];if (num2.getCategory() != ObjectInspector.Category.PRIMITIVE) {throw new UDFArgumentException("第2個參數(shù)不是基本數(shù)據(jù)類型");}// 第2個參數(shù)類型判斷PrimitiveObjectInspector temp2 = (PrimitiveObjectInspector) num2;if (PrimitiveObjectInspector.PrimitiveCategory.INT != temp2.getPrimitiveCategory()) {throw new UDFArgumentException("第2個參數(shù)應(yīng)為INT類型");}// 3.設(shè)置函數(shù)返回值類型（返回一個鍵值對數(shù)據(jù)）ArrayList<String> structFieldNames = new ArrayList<>();ArrayList<ObjectInspector> structFieldObjectInspectors = new ArrayList<>();structFieldNames.add("result");structFieldObjectInspectors.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);return ObjectInspectorFactory.getStandardStructObjectInspector(structFieldNames, structFieldObjectInspectors);}@Overridepublic Object evaluate(DeferredObject[] deferredObjects) throws HiveException {// 完成兩數(shù)相加的邏輯計(jì)算ArrayList<Integer> arrayList = new ArrayList<>();int num1 = Integer.parseInt(deferredObjects[0].get().toString());int num2 = Integer.parseInt(deferredObjects[1].get().toString());arrayList.add(num1 + num2);return arrayList;}@Overridepublic String getDisplayString(String[] strings) {return getStandardDisplayString("AddTestReturnList", strings);}}

（退出當(dāng)前與 Hive 的連接，建立新的連接，刷新緩存）

同樣的，打包上傳到 HDFS 上進(jìn)行注冊：

create function AddTestReturnList as 'AddTestReturnList' using jar 'hdfs://hadoop201:8020/test/MyUDF-1.0-SNAPSHOT-jar-with-dependencies.jar';

此時，可能會發(fā)生報錯，這是由于我們之前已經(jīng)加載過該 JAR 包了，再次加載時 Hive 會拋出異常，我們可以通過下面的語句進(jìn)行調(diào)整：

-- 關(guān)閉向量化查詢
set hive.vectorized.execution.enabled=false;

重新注冊即可。

進(jìn)行測試：

select AddTestReturnList(1,2);

計(jì)算結(jié)果如下：

是不是輕松拿捏了~