学习笔记Hive—— 自定义函数

Posted 2021-10-24 别呀

一、自定义函数简介

1.1、函数类型

UDF：用户定义函数
UDF操作作用于单个数据行，并且产生一个数据行作为输出。大多数函数都属于这一类（比如数学函数和字符串函数）

UDAF：用户定义聚集函数
UDAF 接受多个输入数据行，并产生一个输出数据行。像COUNT和MAX这样的函数就是聚集函数。

UDTF：用户定义表生成函数
UDTF 操作作用于单个数据行，并且产生多个数据行。比如explode。

---

二、自定义UDF

定义一个udf，实现根据输入的日期，输出一个时段， 2:00-5:00凌晨，5:00-12:00为上午，12:00-14:00为中午，14:00-17:00为下午，17:00-19:00为傍晚，19:00-23:00为晚上，23:00-2:00为深夜

• 继承UDF
• 定义evaluate方法
  
  测试数据：

2019-12-12 10:38:26
2019-12-12 17:00:00
2016-01-22 19:23:22
2018-04-02 03:12:00

2.1、步骤：创建临时函数或永久函数

1、创建临时函数
1.上传自定义udf的jar到Linux （将自定义函数打包成datetime.jar并长传到/opt下）
2.在Hive CLI执行：add jar /opt/dateudf.jar;
3.在Hive CLI执行：create temporary function datetotime as 'demo.DateUDF';
（注意：创建临时函数只有退出hive就不能再使用）

2、创建永久函数
1.把自定义函数的jar上传到hdfs中
2.创建永久函数：create function datetotime as 'demo.DateUDF' using jar 'hdfs://master:8020/user/root/dateudf.jar’;

---

2.2、删除自定义函数

删除函数：drop [temporary] function [if exists] [dbname.]函数名;

例：删除永久函数（临时的就没必要了，因为一退出就没了）

---

三、自定义UDAF

求平均数
1.需继承UDAF；
2.内部静态类需继承UDAFEvaluator抽象类，重写方法init()，iterate()，terminatePartial()，merge()，terminate()。

init() 初始化 一般负责初始化内部字段,通常初始化用来存放最终结果的变量
iterate() 每次都会对一个新的值进行聚合计算时都调用该方法,一般会根据计算结果更新用来存放最终结果的变量,如果计算正确或者输入值合法就返回true
terminatePartial() 这个方法直译过来是"终止部分",部分聚合结果的时候调用该方法 必须返回一个封装了聚合计算当前状态的对象,类似于 MapReduce的combiner 
merge() 接受来自 terminatePartial的返回结果,进行合并,hive合并两部分聚合的时候回调用这个方法
terminate() 终止方法 返回最终聚合函数结果

测试数据：

1,700,1200
2,1201,1400
3,1401,2000
4,2001,3000
5,3001,9999

自定义函数：

package demo;

import org.apache.hadoop.hive.ql.exec.UDAF;
import org.apache.hadoop.hive.ql.exec.UDAFEvaluator;

public class UDAFAverage extends UDAF{
	public static class AvgStat{
		private int mcount;
		private double msum;
	}
public static class AvgEvaluator implements UDAFAverage{
	AvgStat avgStat;
	public AvgEvaluator() {
		super();
		avgStat=new AvgStat();
		init();
}
	@Override
	public void init() {
		avgStat.mcount=0;
		avgStat.msum=0;
	
	}
	/**
	 * 介绍原始数据并进行内部轮转
	 * @return
	 */
	public boolean iterate(Double o) {
		if(o!=null) {
			avgStat.mcount++;
			avgStat.msum+=o;
		}
		return true;
	}
	/**
	 * 接收iterate遍历结束后的输出结果进行初次聚集，类似Combiner
	 * @return
	 */
	public AvgStat terminatePartial() {
		if(avgStat.mcount==0)
		return null;
		else return avgStat;

	}
	public boolean merge(AvgStat otheravgState) {
		if(otheravgState!=null) {
		avgStat.mcount+=otheravgState.mcount;
		avgStat.msum+=otheravgState.msum;
		}
		return true;

	}
	/**
	         * terminate返回最终的聚集函数结果 * * @return
	         */
        public Double terminate() {
	        if(avgStat.mcount==0)
	        	return null;
	        else return avgStat.msum/avgStat.mcount;     
	        }
	}
}

接下来就是创建临时函数或永久函数了。。。

---

四、自定义UDTF

定义一个UDTF，可以将“商品1：价格1，商品2：价格2”格式的一列数据解析成商品，价格两个字段

1.继承GenericUDTF，实现initialize, process, close三个方法
2.其中initialize方法主要是判断输入类型并确定返回的字段类型。
3.process方法对udft函数输入的每一行进行操作，通过调用forward方法返回一行或多行数据。
4.close方法在process调用结束后调用，用于进行其它一些额外操作，只执行一次。

测试数据：

shop1:20,shop2:30
shop3:40,shop4:30,shop5:10

自定义函数：

import java.util.ArrayList;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
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.PrimitiveObjectInspector.PrimitiveCategory;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

public class UDTFExplode extends GenericUDTF{

/**
 * 进行输入类型判断，定义输出字段和类型
 * 辅助类objectInspector帮助使用者访问需要序列化或者反序列化的对象
 */
@Override
	public StructObjectInspector initialize(ObjectInspector[] argOIs) throws UDFArgumentException {
		if(argOIs.length!=1) {
			throw new UDFArgumentLengthException("UDTFExplode takes only one argument");
		}
		if(argOIs[0].getCategory()!=ObjectInspector.Category.PRIMITIVE) {
			throw new UDFArgumentException("UDTFExplode takes string as a parameter");
		}
		ArrayList<String> fieldNames = new ArrayList<String>();  
	    ArrayList<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>();
	    fieldNames.add("shop");
	    fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
	    fieldNames.add("volume");
	    fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
	    return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames,fieldOIs);
	}

@Override
	public void close() throws HiveException {
	// TODO Auto-generated method stub
	
	}
/**
 * 处理输入的数据：商品1：销量1，商品2：销量2....
 * 返回多行，每行新增两个字段shop，volumn
 */
@Override
	public void process(Object[] arg0) throws HiveException {
		String[] input = arg0[0].toString().split(",");
		for(String shop_vol:input) {
				String[] input_split = shop_vol.split(":");
				forward(input_split);
		}
	
	}
}

接下来就是创建临时函数或永久函数了。。。

---

4.1、UDTF使用方法

1、直接在select中使用
（udtfexplode：自定义函数）

select udtfexplode(datetime) as (shop,volume) from test;

2、和lateral view一起使用，执行过程相当于单独执行了两次抽取，然后合并到一个表里。

select datetime,t.shop,t.volume from test lateral view udtfexplode(datetime) t as shop,volume;