hive wiki -UDTF

1. UDTF介绍

UDTF(User-Defined Table-Generating Functions)  用来解决 输入一行输出多行(On-to-many maping) 的需求。

2. 编写自己需要的UDTF

继承org.apache.hadoop.hive.ql.udf.generic.GenericUDTF。
实现initialize, process, close三个方法
UDTF首先会调用initialize方法，此方法返回UDTF的返回行的信息（返回个数，类型）。初始化完成后，会调用process方法，对传入的参数进行处理，可以通过forword()方法把结果返回。最后close()方法调用，对需要清理的方法进行清理。
下面是我写的一个UDTF解析json格式，比较纠结的是这个字段很多时候不满足json的定义，有｛｝值，Null一些情况，做了很多判断，不知道页面展示怎么通过的

package com.taobao.voc.hive.udtf;

public class CopyOfUDTFJson2Rows extends GenericUDTF {

	@Override
	public StructObjectInspector initialize(ObjectInspector[] args) throws UDFArgumentException {

		if (args.length != 1 && args.length != 2) {
			throw new UDFArgumentLengthException("UDTFSplitValue takes only one or two argument");
		}

		if (args[0].getCategory() != ObjectInspector.Category.PRIMITIVE) {
			throw new UDFArgumentException("UDTFSplitValue takes string as a parameter");
		}

		ArrayList<String> fieldNames = new ArrayList<String>();
		ArrayList<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>();

		fieldNames.add("col");
		fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);

		return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames, fieldOIs);
	}

	public void process(Object[] args) throws HiveException {
		JSONObject json;
		try {
			json = new JSONObject(args[0].toString());
			if (json.has("bzwd")) {
				String bzwd = json.getString("bzwd");
				bzwd = new JSONObject(bzwd).getString("data");
				JSONObject asks = new JSONObject(bzwd);
				String result = "";
				for (int i = 0; i < asks.getJSONArray("child").length(); i++) {
					result = result + getAskAnswer(asks.getJSONArray("child").getJSONObject(i), 1, (i + 1) + "");
				}
				String[] split = result.split("\n");
				System.out.println(Thread.currentThread().getName() + " " + result);
				for (int i = 0; i < split.length; i++) {
					String[] temp = { split[i] };
					forward(temp);
				}
			} else {
				forward(new String[] { "非标准化问题", "非标准化问题", "非标准化问题", "非标准化问题" });
			}
		} catch (JSONException e) {
			e.printStackTrace();
		}

	}

	/**
	 * @param pNode
	 * json数组 level 问答层级 line 问和答属于第几条线路
	 * */
	public String getAskAnswer(JSONObject pNode, int level, String sLine) throws JSONException {
		String final_result = "";
		if (!pNode.toString().isEmpty() && pNode.has("value") && pNode.has("index")) {
			final_result = final_result + "p_id:" + pLine(sLine) + ";s_id:" + sLine + ";level:" + level + ";ask_id:"
					+ pNode.get("index") + ";answer_id:" + pNode.get("value") + "\n";
		}
		// 子节点有子节点，并且子节点是有效的答案（即value字段有值）
		if (pNode.has("child") && pNode.has("value")) {
			System.out.println(pNode.get("value"));
			for (int j = 0; j < pNode.getJSONArray("child").length(); j++) {
				if (ifContinue(pNode)) {
					final_result = final_result
							+ getAskAnswer(pNode.getJSONArray("child").getJSONObject(j), level + 1, sLine(sLine, j));
				}
			}
		}
		return final_result;
	}
//几个判断节点是否为空和获取p_id的函数省略
}

UDTF有两种使用方法，一种直接放到select后面，一种和lateral view一起使用。

输入格式为JSON
添加jar
add jar /home/taobao/dw_hive/hivelets/smoking/ext/tsa/hivesql/udf/Json2rows.jar;
CREATE TEMPORARY FUNCTION jrow  AS 'com.taobao.voc.hive.udtf.UDTFJson2Rows';
1：直接select中使用：
select jrow(ext_attrs) as format_memo from s_tpp_case_universal;
2：和lateral view一起使用：
select id,format_memo,gmt_create,gmt_modified from s_tpp_case_universal  lateral view jrow(ext_attrs) b as format_memo;
结果：
p_id:0;s_id:1;level:1;ask_id:dpxxwh1;answer_id:ppxgwt2
p_id:1;s_id:1.1;level:2;ask_id:ppxgwt1;answer_id:pptj2
p_id:1.1;s_id:1.1.1;level:3;ask_id:pptj1;answer_id:pptjtjh2
p_id:1.1.1;s_id:1.1.1.1;level:4;ask_id:pptjh1;answer_id:yyshjg2
p_id:1.1.1.1;s_id:1.1.1.1.1;level:5;ask_id:yyshjg1;answer_id:pptjbtg2
p_id:1.1.1.1.1;s_id:1.1.1.1.1.1;level:6;ask_id:shjgbtg1;answer_id:shcw2

评论

2 楼 安铁辉 2013-05-15  

sarahtest 写道

GREAT，HIVE真心强大啊

是的，hive+函数,确实很强大

1 楼 sarahtest 2013-03-28  

GREAT，HIVE真心强大啊