Hive学习——用户自定义函数开发

小鱼儿 2022-12-15 11:27 214阅读 0赞

# 一、前言 #

hive自定义函数分为一些三种：

*  UDF：用户自定义标量函数，输入输出为一对一，如内置的upper函数；
 *  UDAF：用户自定义聚合函数，输入输出为多对一，如内置的sum函数；
 *  UDTF：用户自定义表值函数，输入输出为一对多，如内置的explode函数；

# 二、自定义函数开发步骤 #

不管是哪种自定义函数，hive用户自定义函数从开发到使用都需要大致经过下面几个步骤：

1.  编码实现类，继承相应的UDF类（不同类型的函数继承类不一样），并在相应的方法中实现业务处理逻辑；
2.  打包上传到hive安装目录的lib目录下；
3.  加载jar包，hive重启会自动加载lib下的所有jar包；也可以手动加载jar包，cli命令如下：  
                             add jar /usr/local/apache-hive-3.1.2-bin/lib/hadoop-test-1.0-SNAPSHOT.jar;
4.  创建函数，分为永久函数和临时函数，temporary表示为临时函数，tuomin是函数名，后面引号里为实现类的全限定名，cli命令如下：  
                             create temporary function tuomin as 'cn.lsh.hive.udf.TuoMin';
5.  查看函数信息  
                             desc function extended tuomin;
6.  不需要了可以删除函数  
                             drop temporary function if exists tuomin;

# 三、自定义函数实现样例 #

## 1、UDF ##

UDF函数实现类需要继承GenericUDF类，并重写实现其一下三个方法：

1.  initialize：初始化方法，可以进行参数校验，对象实例化等，并定义evaluate返回类型；
2.  evaluate：计算方法，实现具体的业务逻辑，对入参进行转换，此方法最好返回hadoop标准输出类型如Text，避免出现强制类型转换异常问题；
3.  getDisplayString：显示函数的帮助信息。

样例，姓名脱敏：

public class TuoMin extends GenericUDF {
    	Logger log = LoggerFactory.getLogger(TuoMin.class);
    
    	@Override
    	public ObjectInspector initialize(ObjectInspector[] objectInspectors) throws UDFArgumentException {
    		//初始化，可以进行参数校验，对象实例化等等，定义evaluate返回类型
    		return PrimitiveObjectInspectorFactory.writableStringObjectInspector;
    	}
    
    	@Override
    	public Text evaluate(DeferredObject[] deferredObjects) throws HiveException {
    		//业务逻辑
    		if (deferredObjects.length != 1) {
    			throw new UDFArgumentLengthException("UDF只支持一个入参");
    		}
    		log.info("入参类型：" + deferredObjects[0].get().getClass());
    		String src = deferredObjects[0].get().toString();
    		log.info("入参：" + src);
    		StringBuilder newStr = new StringBuilder(src.charAt(0) + "");
    		if (src.length() > 2) {
    			for (int i = 1; i < src.length(); i++) {
    				if (i == src.length() - 1) {
    					newStr.append(src.charAt(i));
    				} else {
    					newStr.append("*");
    				}
    			}
    		} else {
    			newStr.append("*");
    		}
    		log.info("脱敏后姓名：" + newStr.toString());
    		return new Text(newStr.toString());
    	}
    
    	@Override
    	public String getDisplayString(String[] strings) {
    		//显示函数的帮助信息
    		return "姓名脱敏，只显示姓，如张**";
    	}
    
    	public static void main(String[] args) throws Exception {
    		DeferredObject[] deferredObjects = new DeferredObject[] {new DeferredJavaObject("吴彦祖")};
    		Text text = (Text) new TuoMin().evaluate(deferredObjects);
    		System.out.println(text.toString());
    	}
    }

## 2、UDAF ##

UDAF函数执行实际会跑一个MapReduce任务（UDF和UDTF都不需要），先通过Mapper转换多个入参，最后通过Reducer进行聚合输出。因此UDAF实现复杂一点，分为以下两个部分：

*  函数类：继承AbstractGenericUDAFResolver抽象类，主要实现getEvaluator方法，根据入参选择对应的GenericUDAFEvaluator；
 *  计算类：实现GenericUDAFEvaluator接口，实现UDAFEvaluator接口的init、iterate、terminatePartial、merge、terminate方法。

1.  init：确定各个阶段输入输出参数的数据格式ObjectInspectors；
2.  getNewAggregationBuffer：返回保存数据聚集结果的类；
3.  iterate：map阶段，迭代处理输入sql传过来的列数据；
4.  terminatePartial：map与combiner结束返回结果，得到部分数据聚集结果；
5.  merge：combiner合并map返回的结果，还有reducer合并mapper或combiner返回的结果；
6.  reset：重置聚集结果；
7.  terminate：reducer阶段，输出最终结果。

Hive聚合函数中的几个过程模式：

1.  PARTIAL1（阶段1：map）：init() --> iterate() --> terminatePartial()；
2.  PARTIAL2（阶段2：combine）：init() --> merge() --> terminatePartial()；
3.  FINAL （最终阶段：reduce）：init() --> merge() --> terminate()；
4.  COMPLETE（直接输出阶段：只有map）：init() --> iterate() --> terminate()。

样例，统计所有字符的字符长度总和：

public class ChaCountUDAF extends AbstractGenericUDAFResolver {
    
    	@Override
    	public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters) throws SemanticException {
    		//该方法会根据sql传人的参数数据格式指定调用哪个Evaluator进行处理
    		if (parameters.length != 1) {
    			throw new UDFArgumentTypeException(parameters.length - 1, "Exactly one argument is expected.");
    		}
    		PrimitiveObjectInspector inputOI = (PrimitiveObjectInspector) TypeInfoUtils.getStandardJavaObjectInspectorFromTypeInfo(parameters[0]);
    		if (inputOI.getCategory() != ObjectInspector.Category.PRIMITIVE) {
    			throw new UDFArgumentTypeException(0, "Only primitive type arguments are accepted but " + parameters[0].getTypeName() + " is passed.");
    		}
    		switch (inputOI.getPrimitiveCategory()) {
    		case BYTE:
    		case SHORT:
    		case INT:
    		case LONG:
    		case FLOAT:
    		case DOUBLE:
    		case STRING:
    			//前面的类型都返回ChaCountUDAFEvaluator
    			return new ChaCountUDAFEvaluator();
    		case TIMESTAMP:
    		case BOOLEAN:
    		default:
    			throw new UDFArgumentTypeException(0, "Only primitive type arguments are accepted but " + parameters[0].getTypeName() + " is passed.");
    		}
    	}
    
    	/**
    	 * 自定义静态内部类：数据处理类，继承GenericUDAFEvaluator抽象类
    	 * init(): 确定各个阶段输入输出参数的数据格式ObjectInspectors
    	 * getNewAggregationBuffer(): 保存数据聚集结果的类
    	 * reset(): 重置聚集结果
    	 * iterate(): map阶段，迭代处理输入sql传过来的列数据
    	 * terminatePartial(): map与combiner结束返回结果，得到部分数据聚集结果
    	 * merge(): combiner合并map返回的结果，还有reducer合并mapper或combiner返回的结果
    	 * terminate(): reducer阶段，输出最终结果
    	 */
    	public static class ChaCountUDAFEvaluator extends GenericUDAFEvaluator {
    
    		/**@step1 定义全局输入输出数据的类型OI实例，用于解析输入输出数据*/
    
    		/**
    		 * input For PARTIAL1 and COMPLETE
    		 */
    		PrimitiveObjectInspector inputOI;
    
    		ObjectInspector ouputOI;
    
    		PrimitiveObjectInspector integerOI;
    
    		/**字符总数*/
    		int total = 0;
    
    
    		@Override
    		public ObjectInspector init(Mode m, ObjectInspector[] parameters) throws HiveException {
    			//初始化：对各个模式处理过程，提取输入数据类型OI，返回输出数据类型OI
    			//断言，不满足条件，程序停止执行
    			assert parameters.length == 1;
    			super.init(m, parameters);
    
    			if (m == Mode.PARTIAL1 || m == Mode.COMPLETE) {
    				//map阶段读取sql列，输入为String基础数据格式
    				inputOI = (PrimitiveObjectInspector) parameters[0];
    			} else {
    				//其余阶段，输入为Integer基础数据格式
    				integerOI = (PrimitiveObjectInspector) parameters[0];
    			}
    			// 指定各个阶段输出数据格式都为Integer类型
    			ouputOI = ObjectInspectorFactory.getReflectionObjectInspector(Integer.class, ObjectInspectorFactory.ObjectInspectorOptions.JAVA);
    			return ouputOI;
    		}
    
    		@Override
    		public AggregationBuffer getNewAggregationBuffer() throws HiveException {
    			return new ChaSumAgg();
    		}
    
    		@Override
    		public void reset(AggregationBuffer agg) throws HiveException {
    			ChaSumAgg sumAgg = (ChaSumAgg) agg;
    			sumAgg.sum = 0;
    		}
    
    		@Override
    		public void iterate(AggregationBuffer agg, Object[] parameters) throws HiveException {
    			assert parameters.length == 1;
    			//迭代计算每个字段的字符长度
    			if (parameters[0] != null) {
    				ChaSumAgg sumAgg = (ChaSumAgg) agg;
    				String p1 = PrimitiveObjectInspectorUtils.getString(parameters[0], inputOI);
    				sumAgg.add(p1.length());
    			}
    		}
    
    		@Override
    		public Object terminatePartial(AggregationBuffer agg) throws HiveException {
    			//计算所有字段的总字符长度
    			ChaSumAgg sumAgg = (ChaSumAgg) agg;
    			total += sumAgg.sum;
    			return total;
    		}
    
    		@Override
    		public void merge(AggregationBuffer agg, Object partial) throws HiveException {
    			//combiner合并map返回的结果
    			if (partial != null) {
    				ChaSumAgg sumAgg = (ChaSumAgg) agg;
    				int partialSum = PrimitiveObjectInspectorUtils.getInt(partial, integerOI);
    				sumAgg.add(partialSum);
    			}
    		}
    
    		@Override
    		public Object terminate(AggregationBuffer agg) throws HiveException {
    			//reduce阶段输出最终结果
    			ChaSumAgg sumAgg = (ChaSumAgg) agg;
    			total = sumAgg.sum;
    			return total;
    		}
    
    		/**
    		 * 存储当前字符总数的类
    		 * AggregationBuffer 允许我们保存中间结果，通过定义我们的buffer，我们可以处理任何格式的数据
    		 */
    		static class ChaSumAgg implements AggregationBuffer {
    			int sum = 0;
    
    			void add(int num) {
    				sum += num;
    			}
    		}
    	}
    }

## 3、UDTF ##

UDTF函数实现类需要继承GenericUDTF抽象类，并实现它的下面几个方法：

1.  initialize：设置输出列名，和列类型，返回对象监控器；
2.  process：具体业务处理逻辑，forward输出结果；
3.  close：可以在这里实现一些关流操作。

样例：struct对象转换成表格：

/**
     * 自定义表值函数,将struct类型转换成表格
     */
    public class SplitStructUDTF extends GenericUDTF {
    
    	private transient StructObjectInspector structInputOI;
    
    	/**
    	 * 函数输出结果一行数据
    	 */
    	private transient Object[] forwardStructObj;
    
    	@Override
    	public StructObjectInspector initialize(StructObjectInspector argOI) throws UDFArgumentException {
    		//获取函数的入参域
    		List<? extends StructField> inputFields = argOI.getAllStructFieldRefs();
    		if (inputFields.size() != 1) {
    			throw new UDFArgumentException("splitStruct() takes only one argument");
    		}
    		ObjectInspector udtfInputOI = inputFields.get(0).getFieldObjectInspector();
    
    		//返回的列名
    		List<String> fieldNames = new ArrayList<>();
    		//返回的列值类型
    		List<ObjectInspector> fieldOIs = new ArrayList<>();
    		if (udtfInputOI instanceof StructObjectInspector) {
    			//判断入参是否是struct类型
    			structInputOI = (StructObjectInspector) udtfInputOI;
    			//获得struct中的所有属性
    			List<? extends StructField> structFields = structInputOI.getAllStructFieldRefs();
    			//属性个数与输出值个数对应
    			forwardStructObj = new Object[structFields.size()];
    			for (StructField field : structFields) {
    				//属性作为输出的列名
    				fieldNames.add(field.getFieldName());
    				//申明函数返回值类型
    				fieldOIs.add(field.getFieldObjectInspector());
    				// fieldOIs.add(PrimitiveObjectInspectorFactory.writableStringObjectInspector);
    			}
    			return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);
    		} else {
    			throw new RuntimeException("只能接收Struct入参");
    		}
    	}
    
    	@Override
    	public void process(Object[] args) throws HiveException {
    		if (args != null && args.length > 0) {
    			//获得传入的struct中各属性的值
    			List<Object> values = structInputOI.getStructFieldsDataAsList(args[0]);
    			for (int i = 0; i < values.size(); i++) {
    				forwardStructObj[i] = values.get(i);
    			}
    			//输出一行结果，类似于context.write()
    			forward(forwardStructObj);
    		}
    	}
    
    	@Override
    	public void close() throws HiveException {
    		//可以在这里实现一些关流操作
    	}
    }

select split\_structinfo from struct\_test;                                                         slect split\_struct(info) from test\_struct;

![20201016153409895.png][]                    ![20201016153429508.png][]

[20201016153409895.png]: https://img-blog.csdnimg.cn/20201016153409895.png
[20201016153429508.png]: /images/20221123/667f742100c1436ea595a00b0fdaeb44.png