hive udtf的使用

本文主要是介绍hive udtf的使用，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

原文:http://blog.linezing.com/2011/03/hive%E4%B8%ADudtf%E7%BC%96%E5%86%99%E5%92%8C%E4%BD%BF%E7%94%A8

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()方法调用，对需要清理的方法进行清理。

下面是我写的一个用来切分”key:value;key:value;”这种字符串，返回结果为key, value两个字段。供参考：

import java.util.ArrayList; 

import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF; 
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.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 ExplodeMap extends GenericUDTF{ 

    @Override 
    public void close() throws HiveException { 
        // TODO Auto-generated method stub     
    } 

    @Override 
    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); 

        return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames,fieldOIs); 
    } 

    @Override 
    public void process(Object[] args) throws HiveException { 
        String input = args[0].toString(); 
        String[] test = input.split(";"); 
        for(int i=0; i<test.length; i++) { 
            try { 
                String[] result = test[i].split(":"); 
                forward(result); 
            } catch (Exception e) { 
                continue; 
            } 
        } 
    } 
} 

3. 使用方法

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

1：直接select中使用：select explode_map(properties) as (col1,col2) from src;

不可以添加其他字段使用：select a, explode_map(properties) as (col1,col2) from src
不可以嵌套调用：select explode_map(explode_map(properties)) from src
不可以和group by/cluster by/distribute by/sort by一起使用：select explode_map(properties) as (col1,col2) from src group by col1, col2

2：和lateral view一起使用：select src.id, mytable.col1, mytable.col2 from src lateral view explode_map(properties) mytable as col1, col2;

此方法更为方便日常使用。执行过程相当于单独执行了两次抽取，然后union到一个表里。

4. 参考文档

http://wiki.apache.org/hadoop/Hive/LanguageManual/UDF

http://wiki.apache.org/hadoop/Hive/DeveloperGuide/UDTF

http://www.slideshare.net/pauly1/userdefined-table-generating-functions

5 测试实例参考：

注：自己额外添加

测试数据：

user action 1 2 3

user action 4 5 6 7

user action 8 9 10 11 12

建表：

CREATE external table wftest (user string,action string,objects string) ROW FORMAT SERDE 'org.apache.hadoop.hive.contrib.serde2.RegexSerDe' WITH SERDEPROPERTIES( 'input.regex' ='^([^\\s]*) ([^\\s]*) (.*)$') STORED AS TEXTFILE LOCATION '/user/hdfs/home/wangfeng4/';

hive> desc wftest;

user string from deserializer

action string from deserializer

objects string from deserializer

Time taken: 0.25 seconds

hive> select * from wftest;

user action 1 2 3

user action 4 5 6 7

user action 8 9 10 11 12

NULL NULL NULL

Time taken: 0.245 seconds

hive> select explode(split(objects," ")) as object from wftest;

Total MapReduce jobs = 1

Launching Job 1 out of 1

Number of reduce tasks is set to 0 since there's no reduce operator

655.104: [GC 655.104: [ParNew: 1638400K->25080K(1843200K), 0.0364240 secs] 1638400K->25080K(2150400K), 0.0365450 secs] [Times: user=0.09 sys=0.01, real=0.04 secs]

Starting Job = job_201112221522_3256344, Tracking URL = http://jobtracker.aer.dip.sina.com.cn:50030/jobdetails.jsp?jobid=job_201112221522_3256344

Kill Command = /usr/lib/hadoop/bin/hadoop job -Dmapred.job.tracker=jobtracker.aer.dip.sina.com.cn:8021 -kill job_201112221522_3256344

2012-06-01 16:08:18,085 Stage-1 map = 0%, reduce = 0%

2012-06-01 16:08:21,109 Stage-1 map = 100%, reduce = 0%

2012-06-01 16:08:22,117 Stage-1 map = 100%, reduce = 100%

Ended Job = job_201112221522_3256344