开发HIVE的UDTF自定义函数

[Author]: kwu

UDTF(User-Defined Table-Generating Functions) 用来解决 输入一行输出多行(On-to-many maping) 的需求，开发HIVE的UDTF自定义函数具体步骤如下：

1、继承org.apache.hadoop.hive.ql.udf.generic.GenericUDTF,实现initialize, process, close三个方法。

2、UDTF首先会调用initialize方法，此方法返回UDTF的返回行的信息（返回个数，类型）。

3、初始化完成后，会调用process方法,真正的处理过程在process函数中，在process中，每一次forward()调用产生一行；如果产生多列可以将多个列的值放在一个数组中，然后将该数组传入到forward()函数。

4、最后close()方法调用，对需要清理的方法进行清理。

5、代码实例，实现的功能比较简单，首先按 "\001" 切分，再处理字符串，其中涉及对JSON的处理

package com.hexun.udtf;

import java.util.ArrayList;

import net.sf.json.JSON;
import net.sf.json.JSONSerializer;

import org.apache.commons.beanutils.PropertyUtils;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

public class UDTFDratio extends GenericUDTF {

	public void close() throws HiveException {

	}

	// 返回UDTF的处理行的信息（个数，类型）。
	public StructObjectInspector initialize(ObjectInspector[] args)
			throws UDFArgumentException {
		if (args.length != 1) {
			throw new UDFArgumentLengthException(
					"ExplodeMap takes only one argument");
		}
		if (args[0].getCategory() != ObjectInspector.Category.PRIMITIVE) {
			throw new UDFArgumentException(
					"ExplodeMap takes string as a parameter");
		}
		ArrayList<String> fieldNames = new ArrayList<String>();
		ArrayList<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>();

		fieldNames.add("col1");
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldNames.add("col2");
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldNames.add("col3");
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldNames.add("col4");
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldNames.add("col5");
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldNames.add("col6");
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldNames.add("col7");
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
		fieldNames.add("col8");
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);

		return ObjectInspectorFactory.getStandardStructObjectInspector(
				fieldNames, fieldOIs);
	}

	// 对传入的参数进行处理，可以通过forword()方法返回结果
	public void process(Object[] args) throws HiveException {
		String input = args[0].toString();
		String[] splited = input.split("\001");

		String[] result = new String[8];

		for (int i = 0; i < splited.length; i++) {
			if (i == 0) {
				String head = splited[i];
				String userId = head.substring(0, head.indexOf("_"));
				String cookieId = head.substring(head.indexOf("_") + 1);

				result[0] = userId;
				result[1] = cookieId;
			} else {
				String json = splited[i];
				JSON jo = JSONSerializer.toJSON(json);
				Object o = JSONSerializer.toJava(jo);

				try{
					String sex = PropertyUtils.getProperty(o, "sex").toString();
					result[2] = sex;

					String age = PropertyUtils.getProperty(o, "age").toString();
					result[3] = age;

					String ppt = PropertyUtils.getProperty(o, "ppt").toString();
					result[4] = ppt;

					String degree = PropertyUtils.getProperty(o, "degree").toString();
					result[5] = degree;

					String favor = PropertyUtils.getProperty(o, "favor").toString();
					result[6] = favor;

					String commercial = PropertyUtils.getProperty(o, "commercial").toString();
					result[7] = commercial;

				}catch(Exception e){
					e.printStackTrace();
				}
			}
		}

		forward(result);
	}

}

示例代码涉及的JAR包

6、hive命令行操作，引入UDTF前，需要先加入JSON的依赖包

add jar /opt/softwares/lib/commons-beanutils-1.7.0.jar;
add jar /opt/softwares/lib/commons-collections-3.2.jar;
add jar /opt/softwares/lib/commons-lang-2.4.jar;
add jar /opt/softwares/lib/commons-logging-1.1.3.jar;
add jar /opt/softwares/lib/ezmorph-1.0.3.jar;
add jar /opt/softwares/lib/json-lib-2.2.3-jdk15.jar;
add jar /opt/softwares/UDF.jar;
create temporary function explode_map3 as 'com.hexun.udtf.UDTFDratio';

insert into table stage.dratio PARTITION (day='${yesterday}') select explode_map3(datadratio) as (col1,col2,col3,col4,col5,col6,col7,col8) from stage.dratio_tmp;