UDTF

Posted 2020-06-12

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

编写自己的UDTF：

1.继承org.apache.hadoop.hive.ql.udf.generic.GenericUDTF。

2.实现initialize()，process()，close()三个方法。

3.UDTF首先会调用initialize()方法，此方法返回UDTF的返回行的信息(返回个数，类型)。

4.初始化完成后会调用process()方法，对传入的参数进行处理，可以通过forward()方法把结果返回。

5.最后调用close()对需要清理的方法进行清理。

 

示例：使用UDTF对"Key：Value"这种字符串进行切分，返回结果为Key，Value两个字段。

 

[java] view plain copy

 

1. import java.util.ArrayList;  
2.   
3.  import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;  
4.  import org.apache.hadoop.hive.ql.exec.UDFArgumentException;  
5.  import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;  
6.  import org.apache.hadoop.hive.ql.metadata.HiveException;  
7.  import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;  
8.  import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;  
9.  import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;  
10.  import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;  
11.   
12.  public class ExplodeMap extends GenericUDTF{  
13.   
14.      @Override  
15.      public void close() throws HiveException {  
16.          // TODO Auto-generated method stub      
17.      }  
18.   
19.      @Override  
20.      public StructObjectInspector initialize(ObjectInspector[] args)  
21.              throws UDFArgumentException {  
22.          if (args.length != 1) {  
23.              throw new UDFArgumentLengthException("ExplodeMap takes only one argument");  
24.          }  
25.          if (args[0].getCategory() != ObjectInspector.Category.PRIMITIVE) {  
26.              throw new UDFArgumentException("ExplodeMap takes string as a parameter");  
27.          }  
28.   
29.          ArrayList<String> fieldNames = new ArrayList<String>();  
30.          ArrayList<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>();  
31.          fieldNames.add("col1");  
32.          fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);  
33.          fieldNames.add("col2");  
34.          fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);  
35.   
36.          return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames,fieldOIs);  
37.      }  
38.   
39.      @Override  
40.      public void process(Object[] args) throws HiveException {  
41.          String input = args[0].toString();  
42.          String[] test = input.split(";");  
43.          for(int i=0; i<test.length; i++) {  
44.              try {  
45.                  String[] result = test[i].split(":");  
46.                  forward(result);  
47.              } catch (Exception e) {  
48.                  continue;  
49.              }  
50.          }  
51.      }  
52.  }  

1.打包发送到服务器。

 

2.添加到Hive环境中：

 

[java] view plain copy

 

1. hive (hive)> add jar /usr/local/src/udtf.jar<span style="font-family: Arial, Helvetica, sans-serif;">;</span>  
2. Added /usr/local/src/udtf.jar to class path  
3. Added resource: /usr/local/src/udtf.jar  

3.创建临时函数：

[java] view plain copy

 

1. hive (hive)> create temporary function explode_map as ‘com.lixue.udtf.ExplodeMap‘;  
2. OK  
3. Time taken: 0.0080 seconds  

4.查询(UDTF有两种使用方式，一种是直接放到select后面，另外一种是和lateral view一起使用)：

 

 

[java] view plain copy

 

1. hive (hive)> select explode_map(‘name:lavimer;age:23‘) as (col1,col2) from employees;  
2. //MapReduce  
3. OK  
4. col1    col2  
5. name    lavimer  
6. age 23  
7. name    lavimer  
8. age 23  
9. name    lavimer  
10. age 23  

注：不可以添加其他字段使用，如下：

 

 

[java] view plain copy

 

1. select a, explode_map(properties) as (col1,col2) from src  

不可以嵌套调用：

 

 

[java] view plain copy

 

1. select explode_map(explode_map(properties)) from src  

不可以和group by/cluster by/distribute by/sort by一起使用：

 

 

[java] view plain copy

 

1. select explode_map(properties) as (col1,col2) from src group by col1, col2  

可以和lateral view一起使用：

 

 

[java] view plain copy

 

1. hive (hive)> select user.id,employees.col1,employees.col2 from user lateral view explode_map(‘name:lavimer,age:23‘) employees as col1,col2;  
2. //MapReduce...  
3. OK  
4. id  col1    col2  
5. 1   name    lavimer  
6. 1   age 23  
7. 2   name    lavimer  
8. 2   age 23  
9. 3   name    lavimer  
10. 3   age 23  

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