一、自定义函数
当Hive提供的内置函数无法满足你的业务处理需要时，此时就可以考虑使用用户自定义函数（UDF：user-defined function）。
根据用户自定义函数类别分为以下三种：
1、UDF（User-Defined-Function）
一进一出
类似于:lower/upper/reverse
2、UDAF（User-Defined Aggregation Function）
聚集函数，多进一出
类似于：count/max/min
3、UDTF（User-Defined Table-Generating Functions）
一进多出
如lateral view explode()

1.UDF（User-Defined-Function）
首先定义pom.xml

<dependencies>
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-exec</artifactId>
<version>3.1.2</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>3.3.0</version>
</dependency>
</dependencies>
1
2
3
4
5
6
7
8
9
10
11
12
Java代码,测试代码比较简单，DEMO级别，实际根据工作种的业务重写evaluate方法就行

/*
需求：
UDF --> 1行进来 一行出去
15812345678 ---> 158****5678
*/
@SuppressWarnings("all") //压制警告
public class MyUDF extends UDF {
public String evaluate(String phoneStr){
String str1 = phoneStr.substring(0,3); //158
String str2 = phoneStr.substring(7); //5678

return str1 + "****" + str2; //158****5678
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
转临时函数

1、将代码打包
2、将jar包上传到 /export/server/hive/lib目录，并重命名我们的jar包名称
cd /export/server/hive-3.1.2/lib
mv day19_udf-1.0-SNAPSHOT.jar my_udf.jar

3、hive的客户端添加我们的jar包
add jar /export/server/ hive-3.1.2 /lib/my_udf.jar

4、设置函数与我们的自定义函数关联-临时函数
create temporary function my_jiami as 'cn.itcast.udf.MyUDF';

5、使用自定义的临时函数
select my_jiami('15812345678') from test_user;
1
2
3
4
5
6
7
8
9
10
11
12
13
转永久函数

1、将代码打包
2、将jar包上传到/export/server/hive-3.1.2/lib目录，并重命名我们的jar包名称
cd /export/server/hive-3.1.2/lib
mv day19_udf-1.0-SNAPSHOT.jar my_udf.jar

3、把自定义函数的jar上传到hdfs中.
hadoop fs -mkdir /hive_func
hadoop fs -put my_udf.jar /hive_func

4. 创建永久函数
create function my_jiami2 as 'cn.itcast.udf.MyUDF'
using jar 'hdfs://node1:8020/hive_func/my_udf.jar';
5、 验证
select my_jiami2('15812345678') from test_user;


-- 特别强调；
DataGrip对永久函数支持不够友好，如果希望DataGrip能够永远识别永久函数，则需要重启Hadoop，重启Hive的两个RunJar

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
2.UDTF-单列
public class MyUDTF extends GenericUDTF {
private final transient Object[] forwardListObj = new Object[1];

/**
1、该函数只会执行一次，主要完成初始化工作
*/
@Override
public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {
// 该集合用来指定转换之后出来的每一列叫什么名字（列名）
List<String> fieldNames = new ArrayList<>();
fieldNames.add("my_col1");

// 该集合用来指定转换之后出来的每一列是什么类型（列类型） - String
List<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>() ;
fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
//fieldOIs.add(PrimitiveObjectInspectorFactory.javaIntObjectInspector);

return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);

}

/**
* 每转换一行，该方法就会执行一次
* select my_udtf("zookeeper,hadoop,hdfs,hive,MapReduce",",") word;
* select my_udtf("张三:19,李四:20",":",",")
*
*
* @param objects
* @throws HiveException
*/
@Override
public void process(Object[] objects) throws HiveException {
//1:获取原始数据
String args = objects[0].toString(); // "zookeeper,hadoop,hdfs,hive,MapReduce"

//2:获取数据传入的第二个参数，此处为分隔符
String splitKey = objects[1].toString(); //","

//3.将原始数据按照传入的分隔符进行切分
String[] fields = args.split(splitKey); // ["zookeeper","hadoop","hdfs","hive","MapReduce"]

//4:遍历切分后的结果，并写出
for (String field : fields) {
//将每一个单词添加值对象数组
forwardListObj[0] = field;
//forwardListObj[1] = field;
//将对象数组内容写出 ，每调用一次forward，就会多出一行
forward(forwardListObj);
}
}
@Override
public void close() throws HiveException {
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
这里的测试和UDF一模一样
select my_udtf("zookeeper,hadoop,hdfs,hive,MapReduce",",") word;
1
2
3.UDTF-双列
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

import java.util.ArrayList;
import java.util.List;

public class MyUDTF2 extends GenericUDTF {
private final transient Object[] forwardListObj = new Object[2];

/**
1、该函数只会执行一次，主要完成初始化工作
*/
@Override
public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {
// 该集合用来指定转换之后出来的每一列叫什么名字（列名）
List<String> fieldNames = new ArrayList<>();
fieldNames.add("name");
fieldNames.add("age");

// 该集合用来指定转换之后出来的每一列是什么类型（列类型） - String ， Int
List<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>() ;
fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
fieldOIs.add(PrimitiveObjectInspectorFactory.javaIntObjectInspector);

return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);

}

/**
* 每转换一行，该方法就会执行一次
* select my_udtf("张三:19,李四:20,王五:21",",",":")
* @param objects
* @throws HiveException
*/
@Override
public void process(Object[] objects) throws HiveException {
//1:获取原始数据
String line = objects[0].toString(); // "张三:19,李四:20,王五:21"

//2:获取数据传入的第二个参数，此处逗号隔符
String splitKey1 = objects[1].toString(); //","

//3:获取数据传入的第二个参数，此处冒号分隔符
String splitKey2 = objects[2].toString(); //":"

//4:使用逗号将数据切分成一个个的键值对
String[] array1 = line.split(splitKey1); //["张三:19","李四:20","王五:21"]

//5:遍历数组
for (String kv : array1) {
String[] array2 = kv.split(splitKey2); // ["张三",19]
forwardListObj[0] = array2[0]; // "张三"
forwardListObj[1] = Integer.parseInt(array2[1]); //19

forward(forwardListObj);
}

}

@Override
public void close() throws HiveException {

}
}

免责申明：

本文系转载，版权归原作者所有，如若侵权请联系我们进行删除！

《数据治理行业实践白皮书》下载地址：https://fs80.cn/4w2atu

《数栈V6.0产品白皮书》下载地址：https://fs80.cn/cw0iw1

想了解或咨询更多有关袋鼠云大数据产品、行业解决方案、客户案例的朋友，浏览袋鼠云官网：https://www.dtstack.com/?src=bbs

同时，欢迎对大数据开源项目有兴趣的同学加入「袋鼠云开源框架钉钉技术群」，交流最新开源技术信息，群号码：30537511，项目地址：https://github.com/DTStack