介绍
hive的用户自定义聚合函数(UDAF)是一个很好的功能,集成了先进的数据处理。hive有两种UDAF:简单和通用。顾名思义,简单的UDAF,写的相当简单的,但因为使用Java反射导致性能损失,而且有些特性不能使用,如可变长度参数列表。通用UDAF可以使用所有功能,但是UDAF就写的比较复杂,不直观。
本文只介绍通用UDAF。
UDAF是需要在hive的sql语句和group by联合使用,hive的group by对于每个分组,只能返回一条记录,这点和mysql不一样,切记。
UDAF开发概览
开发通用UDAF有两个步骤,第一个是编写resolver类,第二个是编写evaluator类。resolver负责类型检查,操作符重载。evaluator真正实现UDAF的逻辑。通常来说,顶层UDAF类继承org.apache.hadoop.hive.ql.udf.GenericUDAFResolver2,里面编写嵌套类evaluator 实现UDAF的逻辑。
本文以Hive的内置UDAF sum函数的源代码作为示例讲解。
实现 resolver
resolver通常继承org.apache.hadoop.hive.ql.udf.GenericUDAFResolver2,但是我们更建议继承AbstractGenericUDAFResolver,隔离将来hive接口的变化。
GenericUDAFResolver和GenericUDAFResolver2接口的区别是,后面的允许evaluator实现可以访问更多的信息,例如DISTINCT限定符,通配符FUNCTION(*)。
public class GenericUDAFSum extends AbstractGenericUDAFResolver { static final Log LOG = LogFactory.getLog(GenericUDAFSum.class.getName()); @Override public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters) throws SemanticException { // Type-checking goes here! return new GenericUDAFSumLong(); } public static class GenericUDAFSumLong extends GenericUDAFEvaluator { // UDAF logic goes here! } }
这个就是UDAF的代码骨架,第一行创建LOG对象,用来写入警告和错误到hive的log。GenericUDAFResolver只需要重写一个方法:getEvaluator,它根据SQL传入的参数类型,返回正确的evaluator。这里最主要是实现操作符的重载。
getEvaluator的完整代码如下:
public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters) throws SemanticException { if (parameters.length != 1) { throw new UDFArgumentTypeException(parameters.length - 1, "Exactly one argument is expected."); } if (parameters[0].getCategory() != ObjectInspector.Category.PRIMITIVE) { throw new UDFArgumentTypeException(0, "Only primitive type arguments are accepted but " + parameters[0].getTypeName() + " is passed."); } switch (((PrimitiveTypeInfo) parameters[0]).getPrimitiveCategory()) { case BYTE: case SHORT: case INT: case LONG: case TIMESTAMP: return new GenericUDAFSumLong(); case FLOAT: case DOUBLE: case STRING: return new GenericUDAFSumDouble(); case BOOLEAN: default: throw new UDFArgumentTypeException(0, "Only numeric or string type arguments are accepted but " + parameters[0].getTypeName() + " is passed."); }
这里做了类型检查,如果不是原生类型(即符合类型,array,map此类),则抛出异常,还实现了操作符重载,对于整数类型,使用GenericUDAFSumLong实现UDAF的逻辑,对于浮点类型,使用GenericUDAFSumDouble实现UDAF的逻辑。
实现evaluator
所有evaluators必须继承抽象类org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator。子类必须实现它的一些抽象方法,实现UDAF的逻辑。
GenericUDAFEvaluator有一个嵌套类Mode,这个类很重要,它表示了udaf在mapreduce的各个阶段,理解Mode的含义,就可以理解了hive的UDAF的运行流程。
public static enum Mode { /** * PARTIAL1: 这个是mapreduce的map阶段:从原始数据到部分数据聚合 * 将会调用iterate()和terminatePartial() */ PARTIAL1, /** * PARTIAL2: 这个是mapreduce的map端的Combiner阶段,负责在map端合并map的数据::从部分数据聚合到部分数据聚合: * 将会调用merge() 和 terminatePartial() */ PARTIAL2, /** * FINAL: mapreduce的reduce阶段:从部分数据的聚合到完全聚合 * 将会调用merge()和terminate() */ FINAL, /** * COMPLETE: 如果出现了这个阶段,表示mapreduce只有map,没有reduce,所以map端就直接出结果了:从原始数据直接到完全聚合 * 将会调用 iterate()和terminate() */ COMPLETE };
一般情况下,完整的UDAF逻辑是一个mapreduce过程,如果有mapper和reducer,就会经历PARTIAL1(mapper),FINAL(reducer),如果还有combiner,那就会经历PARTIAL1(mapper),PARTIAL2(combiner),FINAL(reducer)。
而有一些情况下的mapreduce,只有mapper,而没有reducer,所以就会只有COMPLETE阶段,这个阶段直接输入原始数据,出结果。
下面以GenericUDAFSumLong的evaluator实现讲解
public static class GenericUDAFSumLong extends GenericUDAFEvaluator { private PrimitiveObjectInspector inputOI; private LongWritable result; //这个方法返回了UDAF的返回类型,这里确定了sum自定义函数的返回类型是Long类型 @Override public ObjectInspector init(Mode m, ObjectInspector[] parameters) throws HiveException { assert (parameters.length == 1); super.init(m, parameters); result = new LongWritable(0); inputOI = (PrimitiveObjectInspector) parameters[0]; return PrimitiveObjectInspectorFactory.writableLongObjectInspector; } /** 存储sum的值的类 */ static class SumLongAgg implements AggregationBuffer { boolean empty; long sum; } //创建新的聚合计算的需要的内存,用来存储mapper,combiner,reducer运算过程中的相加总和。 @Override public AggregationBuffer getNewAggregationBuffer() throws HiveException { SumLongAgg result = new SumLongAgg(); reset(result); return result; } //mapreduce支持mapper和reducer的重用,所以为了兼容,也需要做内存的重用。 @Override public void reset(AggregationBuffer agg) throws HiveException { SumLongAgg myagg = (SumLongAgg) agg; myagg.empty = true; myagg.sum = 0; } private boolean warned = false; //map阶段调用,只要把保存当前和的对象agg,再加上输入的参数,就可以了。 @Override public void iterate(AggregationBuffer agg, Object[] parameters) throws HiveException { assert (parameters.length == 1); try { merge(agg, parameters[0]); } catch (NumberFormatException e) { if (!warned) { warned = true; LOG.warn(getClass().getSimpleName() + " " + StringUtils.stringifyException(e)); } } } //mapper结束要返回的结果,还有combiner结束返回的结果 @Override public Object terminatePartial(AggregationBuffer agg) throws HiveException { return terminate(agg); } //combiner合并map返回的结果,还有reducer合并mapper或combiner返回的结果。 @Override public void merge(AggregationBuffer agg, Object partial) throws HiveException { if (partial != null) { SumLongAgg myagg = (SumLongAgg) agg; myagg.sum += PrimitiveObjectInspectorUtils.getLong(partial, inputOI); myagg.empty = false; } } //reducer返回结果,或者是只有mapper,没有reducer时,在mapper端返回结果。 @Override public Object terminate(AggregationBuffer agg) throws HiveException { SumLongAgg myagg = (SumLongAgg) agg; if (myagg.empty) { return null; } result.set(myagg.sum); return result; } }
相关推荐
hive udaf 实现按位取与或 hive udaf 实现按位取与或 hive udaf 实现按位取与或
A custom UDAF to group oncatenates all arguments from different rows into a single string.
Hive从入门到精通,适合想学hive的小白看
Hive从入门到精通需要软件及配置
Hive入门与实战 PDF
Hive编程指南+HIVE从入门到精通+Hive高级编程+Apache Oozie
hive编程入门课程
个人 Hive UDAF 有一堆 Hive UDAF(用户定义的聚合函数)不在标准 Hive 分布中,因为它们可能会导致大型数据集的 OOM。 要使用它们,您需要加载 jar 文件,然后为每个要使用的函数创建一个临时函数: ADD JAR target...
Hive入门与实战
HIVE从入门到精通,Hive是基于Hadoop(HDFS, MapReduce)的一个数据仓库工具,可以将结构化的数据文件映射为一张数据库表,并提供类SQL查询功能。本质是将SQL转换为MapReduce程序。
hive入门级详解,包括数仓与传统数据库的比对,hive的存储结构与存储原理,分区分桶、hql如何转换成mapreduce、UDF自定义函数等
Hive入门经典教程 Apache Hive 淘宝团队验证
Hive本身建立在Hadoop的体系结构上,可以将结构化的数据文件映射为一张数据库表,并提供完整的sql查询功能,可以将sql语句转换为MapReduce任务进行。并按照该计划生成MapReduce任务后交给Hadoop集群处理。
大数据系列-Hive入门与实战.pptx
Hive是基于Hadoop的一个数据仓库工具,可以将结构化的数据文件映射为一张数据库表,并提供类SQL查询...Hive入门与实战.pdf hive函数大全.doc Hive用户手册中文版.pdf Hive用户指南-V1.1.doc 喜欢的朋友可以下载学习。
hive从入门到放弃(六)——常用文件存储格式.doc
hive入门,介绍hive相关简单知识和例子。
Hive是基于Hadoop的一个数据仓库工具,可以将结构化的数据文件映射为一张数据库表,并提供类SQL查询功能。
hive入门文档笔记,安装配置。创建删除表,加载数据。
Hive入门基础知识.pdf