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
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;
