博客

1、自定义函数的分类

• UDF：一进一出 （例：你输入多少行就给你返回多少行）
• UDTF：一进多出 （例：你输入一行数据给你返回多行）
• UDAF：多进一出

2、UDF案例实现

分析：上面的需求是一进一出的场景，所以我们需要编写UDF自定义函数。

• 代码实现
  (1) 继承org.apache.hadoop.hive.ql.exec.UDF，在里面实现 evaluate方法，在 evaluate 方法中实现自己的逻辑。

package com.shujia.function_hive;
import org.apache.hadoop.hive.ql.exec.UDF;

public class MyUdf extends UDF {
	//实现evaluate方法
	public String evaluate(String col){
	// 张三 ==》 #张三$
	String s = "#"+col+"$";
	return s;
	}
}

(2) 将我们写好代码打包上传Linux 虚拟机
(3) 在 hive shell 中，使用 add jar 路径将 jar 包作为资源添加到 hive 环境中

hive (shuangqian)> add jar /usr/local/soft/data/shuangqian-1.0-SNAPSHOT.jar;

(4) 注册临时函数 fxxxl是你起得函数名，myudf:是我们的类名

hive (shuangqian)> create temporary function fxxxl as 'com.shujia.function_hive.MyUdf';

(5) 使用这个函数

select fxxxl(name) from student limit 10;

(6) 使用函数名处理数据

select fxx1(name) as fxx_name from students limit 10;
#施笑槐$
#吕金鹏$
#单乐蕊$
#葛德曜$
#宣谷芹$
#边昂雄$
#尚孤风$
#符半双$
#沈德昌$
#羿彦昌$

3、UDTF案例实现

(1) 需求

hive(default)> select myudtf("hello,world,hadoop,hive", ",");
hello
world
hadoop
hive

(2) 自定义UDTF函数实现案例一

package com.shujia.function_hive;

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.PrimitiveObjectInspector
Factory;
import java.util.ArrayList;

public class MyUdtf extends GenericUDTF {
	//自定义一个集合
	private ArrayList<String> outlist = new ArrayList<>();
	
		@Override
		public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {
			//定义数据的输出类型和字段的类型
			ArrayList<String> fieldname = new ArrayList<>();
			ArrayList<ObjectInspector> fieldOIs = new ArrayList<>();
			//添加输出列名到集合中(有几个输出列就添加几个列名)
			fieldname.add("lineToWord");
			//添加数据的输出类型到集合中
			fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
			//返回值
			return ObjectInspectorFactory.getStandardStructObjectInspector(fieldname, fieldOIs);
		}
	
		//主要逻辑在这个地方处理
		@Override
		public void process(Object[] args) throws HiveException {
			//获取原始数据，将原始数据转变成string类型
			String s = args[0].toString();
			//获取数据传入的第二个参数，此处为分隔符，将原始数据按照传入的分隔符进行切分
			String[] split = s.split(args[1].toString());
			//遍历输出结果
			for (String s1 : split) {
			//集合为复用的，首先清空集合
			outlist.clear();
			//添加数据到集合
			outlist.add(s1);
			//将集合内容写出
			forward(outlist);
		}
		
		@Override
		public void close() throws HiveException {
		}
}

(3) 自定义UDTF函数实现案例二

• 需求

字段： id,col1,col2,col3,col4,col5,col6,col7,col8,col9,col10,col11,col12 共 13 列
数据： 
a,1,2,3,4,5,6,7,8,9,10,11,12
b,11,12,13,14,15,16,17,18,19,20,21,22
c,21,22,23,24,25,26,27,28,29,30,31,32
转成 3 列：id,hours,value 例如：
a,1,2,3,4,5,6,7,8,9,10,11,12
a,0 时,1
a,2 时,2
a,4 时,3
a,6 时,4

• 创建表并导入数据

create table udtfData(
	id string,
	col1 string,
	col2 string,
	col3 string,
	col4 string,
	col5 string,
	col6 string,
	col7 string,
	col8 string,
	col9 string,
	col10 string,
	col11 string,
	col12 string
)row format delimited fields terminated by ',';

• 代码实现

package com.shujia.function_hive;
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.PrimitiveObjectInspector
Factory;
import java.util.ArrayList;

public class MyUdtf2 extends GenericUDTF {
	@Override
	public StructObjectInspector initialize(StructObjectInspector argOIs) throws
	UDFArgumentException {
		//定义数据的输出类型和字段类型
		ArrayList<String> fieldnames = new ArrayList<>();
		ArrayList<ObjectInspector> fieldobj = new ArrayList<>();

		//注意：虽然最终结果输出三列，但这里只添加了两列，
		//因为这个UDTF函数只是对后面两列进行处理，而不处理id列
		fieldnames.add("col1"); 
		fieldobj.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldnames.add("col2");
		fieldobj.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		return ObjectInspectorFactory.getStandardStructObjectInspector(fieldnames, fieldobj);
	}

	//处理逻辑
	@Override
	public void process(Object[] objects) throws HiveException {
		int hour=0;
		for (Object object : objects) {
		hour = hour+1;
		String col = object.toString();
		ArrayList<String>cols = new ArrayList<>();
		//把数据添加到数组里面去
		cols.add(hour+"时");
		cols.add(col);
		forward(cols);
		}
	}
	
	@Override
	public void close() throws HiveException {
	}
}

• 添加jar 资源:

add jar /usr/local/soft/HiveUDF2-1.0.jar;

• 注册udtf 函数：

create temporary function my_udtf as 'MyUDTF';

• SQL:

select id,hours,value from udtfData lateral view my_udtf(col1,col2,col3,col4,col5,col6,col7,col8,col9,col10,col11,col12) t as hours,value ;