实例

1. @Description(name = "letters", value = "_FUNC_(expr) - 返回该列中所有字符串的字符总数")  
2. public class TotalNumOfLettersGenericUDAF extends AbstractGenericUDAFResolver {  
3.   
4.     @Override  
5.     public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters)  
6.             throws SemanticException {  
7.         if (parameters.length != 1) {  
8.             throw new UDFArgumentTypeException(parameters.length - 1,  
9.                     "Exactly one argument is expected.");  
10.         }  
11.           
12.         ObjectInspector oi = TypeInfoUtils.getStandardJavaObjectInspectorFromTypeInfo(parameters[0]);  
13.           
14.         if (oi.getCategory() != ObjectInspector.Category.PRIMITIVE){  
15.             throw new UDFArgumentTypeException(0,  
16.                             "Argument must be PRIMITIVE, but "  
17.                             + oi.getCategory().name()  
18.                             + " was passed.");  
19.         }  
20.           
21.         PrimitiveObjectInspector inputOI = (PrimitiveObjectInspector) oi;  
22.           
23.         if (inputOI.getPrimitiveCategory() != PrimitiveObjectInspector.PrimitiveCategory.STRING){  
24.             throw new UDFArgumentTypeException(0,  
25.                             "Argument must be String, but "  
26.                             + inputOI.getPrimitiveCategory().name()  
27.                             + " was passed.");  
28.         }  
29.           
30.         return new TotalNumOfLettersEvaluator();  
31.     }  
32.   
33.     public static class TotalNumOfLettersEvaluator extends GenericUDAFEvaluator {  
34.   
35.         PrimitiveObjectInspector inputOI;  
36.         ObjectInspector outputOI;  
37.         PrimitiveObjectInspector integerOI;  
38.           
39.         int total = 0;  
40.   
41.         @Override  
42.         public ObjectInspector init(Mode m, ObjectInspector[] parameters)  
43.                 throws HiveException {  
44.               
45.             assert (parameters.length == 1);  
46.             super.init(m, parameters);  
47.              
48.              //map阶段读取sql列，输入为String基础数据格式  
49.             if (m == Mode.PARTIAL1 || m == Mode.COMPLETE) {  
50.                 inputOI = (PrimitiveObjectInspector) parameters[0];  
51.             } else {  
52.             //其余阶段，输入为Integer基础数据格式  
53.                 integerOI = (PrimitiveObjectInspector) parameters[0];  
54.             }  
55.   
56.              // 指定各个阶段输出数据格式都为Integer类型  
57.             outputOI = ObjectInspectorFactory.getReflectionObjectInspector(Integer.class,  
58.                     ObjectInspectorOptions.JAVA);  
59.             return outputOI;  
60.   
61.         }  
62.   
63.         /** 
64.          * 存储当前字符总数的类 
65.          */  
66.         static class LetterSumAgg implements AggregationBuffer {  
67.             int sum = 0;  
68.             void add(int num){  
69.                 sum += num;  
70.             }  
71.         }  
72.   
73.         @Override  
74.         public AggregationBuffer getNewAggregationBuffer() throws HiveException {  
75.             LetterSumAgg result = new LetterSumAgg();  
76.             return result;  
77.         }  
78.   
79.         @Override  
80.         public void reset(AggregationBuffer agg) throws HiveException {  
81.             LetterSumAgg myagg = new LetterSumAgg();  
82.         }  
83.           
84.         private boolean warned = false;  
85.   
86.         @Override  
87.         public void iterate(AggregationBuffer agg, Object[] parameters)  
88.                 throws HiveException {  
89.             assert (parameters.length == 1);  
90.             if (parameters[0] != null) {  
91.                 LetterSumAgg myagg = (LetterSumAgg) agg;  
92.                 Object p1 = ((PrimitiveObjectInspector) inputOI).getPrimitiveJavaObject(parameters[0]);  
93.                 myagg.add(String.valueOf(p1).length());  
94.             }  
95.         }  
96.   
97.         @Override  
98.         public Object terminatePartial(AggregationBuffer agg) throws HiveException {  
99.             LetterSumAgg myagg = (LetterSumAgg) agg;  
100.             total += myagg.sum;  
101.             return total;  
102.         }  
103.   
104.         @Override  
105.         public void merge(AggregationBuffer agg, Object partial)  
106.                 throws HiveException {  
107.             if (partial != null) {  
108.                   
109.                 LetterSumAgg myagg1 = (LetterSumAgg) agg;  
110.                   
111.                 Integer partialSum = (Integer) integerOI.getPrimitiveJavaObject(partial);  
112.                   
113.                 LetterSumAgg myagg2 = new LetterSumAgg();  
114.                   
115.                 myagg2.add(partialSum);  
116.                 myagg1.add(myagg2.sum);  
117.             }  
118.         }  
119.   
120.         @Override  
121.         public Object terminate(AggregationBuffer agg) throws HiveException {  
122.             LetterSumAgg myagg = (LetterSumAgg) agg;  
123.             total = myagg.sum;  
124.             return myagg.sum;  
125.         }  
126.   
127.     }  
128. }  

1. /** 
2. * 保存当前字符总数的类 
3. */  
4. static class LetterSumAgg implements AggregationBuffer {  
5.     int sum = 0;  
6.     void add(int num){  
7.         sum += num;  
8.     }  
9. }  

1. // init方法中根据不同的mode指定输出数据的格式objectinspector  
2. if (m == Mode.PARTIAL1 || m == Mode.COMPLETE) {  
3.     inputOI = (PrimitiveObjectInspector) parameters[0];  
4. } else {  
5.     integerOI = (PrimitiveObjectInspector) parameters[0];  
6. }  
7.   
8. // 不同model阶段的输出数据格式  
9. outputOI = ObjectInspectorFactory.getReflectionObjectInspector(Integer.class,  
10.                     ObjectInspectorOptions.JAVA);  

1. public void iterate(AggregationBuffer agg, Object[] parameters)  
2.     throws HiveException {  
3.     ...  
4.     Object p1 = ((PrimitiveObjectInspector) inputOI).getPrimitiveJavaObject(parameters[0]);  
5.     myagg.add(String.valueOf(p1).length());  
6.     }  
7. }  

1. public void merge(AggregationBuffer agg, Object partial)  
2.         throws HiveException {  
3.     if (partial != null) {  
4.                   
5.         LetterSumAgg myagg1 = (LetterSumAgg) agg;  
6.                   
7.         Integer partialSum = (Integer) integerOI.getPrimitiveJavaObject(partial);  
8.                   
9.         LetterSumAgg myagg2 = new LetterSumAgg();  
10.                   
11.         myagg2.add(partialSum);  
12.         myagg1.add(myagg2.sum);  
13.     }  
14. }  

1. public Object terminate(AggregationBuffer agg) throws HiveException {  
2.     LetterSumAgg myagg = (LetterSumAgg) agg;  
3.     total = myagg.sum;  
4.     return myagg.sum;  
5. }  

资料参考