1. 引言

在前一篇中，解决了Hive表中复杂数据结构平铺化以导入Kylin的问题，但是平铺之后计算广告日志的曝光PV是翻倍的，因为一个用户对应于多个标签。所以，为了计算曝光PV，我们得另外创建视图。 分析需求：

• 每个DSP上的曝光PV，标签覆盖的曝光PV；
• 累计曝光PV，累计标签覆盖曝光PV

相当于cube(dsp, tag) + measure(pv)，HiveQL如下： select dsp, tag, count(*) as pv from ad_view where view = 'view' and day_time between '2016-04-18' and '2016-04-24' group by dsp, tag with cube; 现在问题来了：如何将原始表中的tags array> 转换成有标签（taged）、无标签（empty）呢？显而易见的办法，为字段tags写一个UDF来判断是否有标签。

2. 实战

基本介绍

user-defined function (UDF)包括：

• 对于字段进行转换操作的函数，如round()、abs()、concat()等；
• 聚集函数user-defined aggregate functions (UDAFs)，比如sum()、avg()等；
• 表生成函数user-defined table generating functions (UDTFs)，生成多列或多行数据，比如explode()、inline()等

UDTF的使用在与select语句使用时受到了限制，比如，不能与其他的列组合出现： hive> SELECT name, explode(subordinates) FROM employees; FAILED: Error in semantic analysis: UDTF's are not supported outside the SELECT clause, nor nested in expressions Hive提供LATERAL VIEW关键字，对UDTF的输入进行包装（wrap），如此可以达到列组合的效果： hive> SELECT name, sub > FROM employees > LATERAL VIEW explode(subordinates) subView AS sub;

UDF与GenericUDF

org.apache.hadoop.hive.ql.exec.UDF是字段转换操作的基类，提供对于简单数据类型进行转换操作。在实现转换操作时，需要重写evaluate()方法。较UDF抽象类，org.apache.hadoop.hive.ql.udf.generic.GenericUDF提供更为复杂的处理方法类，包括三个方法：

• initialize(ObjectInspector[] arguments)，检查输入参数的类型、确定返回值的类型；
• evaluate(DeferredObject[] arguments)，字段转换操作的实现函数，其返回值的类型与initialize方法中所指定的返回类型保持一致；
• getDisplayString(String[] children)，给Hadoop任务展示debug信息的。

判断tags array>是否为空标签（EMPTY）的UDF实现如下： @Description(name = "checkTag", value = "_FUNC_(array) - from the input array of struct "+ "returns the TAGED or EMPTY(no tag).", extended = "Example: " + " > SELECT _FUNC_(tags_array) FROM src;") public class CheckTag extends GenericUDF { private ListObjectInspector listOI; public ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentException { if (arguments.length != 1) { throw new UDFArgumentLengthException("only takes 1 arguments: List"); } ObjectInspector a = arguments[0]; if (!(a instanceof ListObjectInspector)) { throw new UDFArgumentException("first argument must be a list / array"); } this.listOI = (ListObjectInspector) a; if(!(listOI.getListElementObjectInspector() instanceof StructObjectInspector)) { throw new UDFArgumentException("first argument must be a list of struct"); } return PrimitiveObjectInspectorFactory.javaStringObjectInspector; } public Object evaluate(DeferredObject[] arguments) throws HiveException { if(listOI == null || listOI.getListLength(arguments[0].get()) == 0) { return "null_field"; } StructObjectInspector structOI = (StructObjectInspector) listOI.getListElementObjectInspector(); String tag = structOI.getStructFieldData(listOI.getListElement(arguments[0].get(), 0), structOI.getStructFieldRef("tag")).toString(); if (listOI.getListLength(arguments[0].get()) == 1 && tag.equals("EMPTY")) { return "EMPTY"; } return "TAGED"; } public String getDisplayString(String[] children) { return "check tag whether is empty"; } } 还需添加依赖： org.apache.hive hive-exec 0.14.0 provided org.apache.hadoop hadoop-common 2.5.0-cdh5.3.2 provided 编译后打成jar包，放在HDFS上，然后add jar即可调用该UDF了： add jar hdfs://path/to/udf-1.0-SNAPSHOT.jar; create temporary function checktag as 'com.hive.udf.CheckTag'; create view if not exists yooshu_view partitioned on (day_time) as select uid, dsp, view, click, checktag(tags) as tag, day_time from ad_base;