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<struct<tag:string,label:string,src:string>> 转换成有标签（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<struct<tag:string,label:string,src:string>>是否为空标签（EMPTY）的UDF实现如下：

@Description(name = "checkTag",value = "_FUNC_(array<struct>) - from the input array of struct "+"returns the TAGED or EMPTY(no tag).",extended = "Example:\n"+ " > 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<T>");}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";}}

还需添加依赖：

<dependency><groupId>org.apache.hive</groupId><artifactId>hive-exec</artifactId><version>0.14.0</version><scope>provided</scope>
</dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>2.5.0-cdh5.3.2</version><scope>provided</scope>
</dependency>

编译后打成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;