[转]HIVE UDF/UDAF/UDTF的Map Reduce代码框架模板

FROM : http://hugh-wangp.iteye.com/blog/1472371

自己写代码时候的利用到的模板

1.必须继承org.apache.hadoop.hive.ql.exec.UDF

2.必须实现evaluate函数，evaluate函数支持重载

---

Java代码  

1. <span style="font-size: x-small;">package com.alibaba.hive.udf;  
2.   
3. import org.apache.hadoop.hive.ql.exec.UDF  
4.   
5. public class helloword extends UDF{  
6.      public String evaluate(){  
7.           return "hello world!";  
8.      }  
9.   
10.      public String evaluate(String str){  
11.           return "hello world: " + str;  
12.      }  
13. }</span>  

 

1.必须继承

     org.apache.hadoop.hive.ql.exec.UDAF(函数类继承)

     org.apache.hadoop.hive.ql.exec.UDAFEvaluator(内部类Evaluator实现UDAFEvaluator接口)

2.Evaluator需要实现 init、iterate、terminatePartial、merge、terminate这几个函数

     init():类似于构造函数，用于UDAF的初始化

     iterate():接收传入的参数，并进行内部的轮转。其返回类型为boolean

     terminatePartial():无参数，其为iterate函数轮转结束后，返回乱转数据，iterate和terminatePartial类似于hadoop的Combiner(iterate--mapper;terminatePartial--reducer)

     merge():接收terminatePartial的返回结果，进行数据merge操作，其返回类型为boolean

     terminate():返回最终的聚集函数结果

Java代码  

1. <span style="font-size: x-small;">package com.alibaba.hive;  
2.   
3. import org.apache.hadoop.hive.ql.exec.UDAF;  
4. import org.apache.hadoop.hive.ql.exec.UDAFEvaluator;  
5.   
6. public class myAVG extends UDAF{  
7.   
8.      public static class avgScore{  
9.           private long pSum;  
10.           private double pCount;  
11.      }  
12.        
13.      public static class AvgEvaluator extends UDAFEvaluator{  
14.           avgScore score;  
15.             
16.           public AvgEvaluator(){  
17.                score = new avgScore();  
18.                init();  
19.           }  
20.             
21.           /* 
22.           *init函数类似于构造函数，用于UDAF的初始化 
23.           */  
24.           public void init(){  
25.                score.pSum = 0;  
26.                score.pCount = 0;  
27.           }  
28.             
29.           /* 
30.           *iterate接收传入的参数，并进行内部的轮转。其返回类型为boolean 
31.           *类似Combiner中的mapper 
32.           */  
33.           public boolean iterate(Double in){  
34.                if(in != null){  
35.                     score.pSum += in;  
36.                     score.pCount ++;  
37.                }  
38.                return true;  
39.           }  
40.             
41.           /* 
42.           *terminatePartial无参数，其为iterate函数轮转结束后，返回轮转数据 
43.           *类似Combiner中的reducer 
44.           */  
45.           public avgScore terminatePartial(){  
46.                return score.pCount == 0 ? null : score;  
47.           }  
48.             
49.           /* 
50.           *merge接收terminatePartial的返回结果，进行数据merge操作，其返回类型为boolean 
51.           */  
52.           public boolean merge(avgScore in){  
53.                if(in != null){  
54.                     score.pSum += in.pSum;  
55.                     score.pCount += in.pCount;  
56.                }  
57.                return true;  
58.           }  
59.             
60.           /* 
61.           *terminate返回最终的聚集函数结果 
62.           */  
63.           public Double terminate(){  
64.                return score.pCount == 0 ? null : Double.valueof(score.pSum/score.pCount);  
65.           }  
66.      }  
67. }</span>  

---

Java代码  

1. <span style="font-size: x-small;"><span style="font-size: xx-small;">public class GenericUDTFExplode extends GenericUDTF {  
2.   
3.   private ListObjectInspector listOI = null;  
4.   
5.   @Override  
6.   public void close() throws HiveException {  
7.   }  
8.   
9.   @Override  
10.   public StructObjectInspector initialize(ObjectInspector[] args) throws UDFArgumentException {  
11.     if (args.length != 1) {  
12.       throw new UDFArgumentException("explode() takes only one argument");  
13.     }  
14.   
15.     if (args[0].getCategory() != ObjectInspector.Category.LIST) {  
16.       throw new UDFArgumentException("explode() takes an array as a parameter");  
17.     }  
18.     listOI = (ListObjectInspector) args[0];  
19.   
20.     ArrayList<String> fieldNames = new ArrayList<String>();  
21.     ArrayList<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>();  
22.     fieldNames.add("col");  
23.     fieldOIs.add(listOI.getListElementObjectInspector());  
24.     return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames,  
25.         fieldOIs);  
26.   }  
27.   
28.   private final Object[] forwardObj = new Object[1];  
29.   
30.   @Override  
31.   public void process(Object[] o) throws HiveException {  
32.     List<?> list = listOI.getList(o[0]);  
33.     if(list == null) {  
34.       return;  
35.     }  
36.     for (Object r : list) {  
37.       forwardObj[0] = r;  
38.       forward(forwardObj);  
39.     }  
40.   }  
41.   
42.   @Override  
43.   public String toString() {  
44.     return "explode";  
45.   }  
46. }</span></span>  

[转]HIVE UDF/UDAF/UDTF的Map Reduce代码框架模板

原文：http://www.cnblogs.com/Athrun/p/4587192.html