Hive自定义UDAF

2017-08-25  本文已影响164人  voidxin

注:本文copy自http://blog.csdn.net/lzm1340458776/article/details/43313919,纯属当笔记使用。

Hive在查询数据时,有些聚类函数框架并没有提供实现,需要用户自定义实现,这称为UDAF(User-Defined Aggregation Function)。UDAF是针对多条记录的,通俗的讲就是多进一出。下面是实现UDAF的步骤:
1:函数类需要继承UDAF类,并且该函数类中必须提供一个实现了UDAFEvaluator接口的内部类。
2.内部类需要实现init()、iterate()、terminatePartial()、merge()、terminate()这几个函数。
2.1:init()函数实现接口UDAFEvaluator中的init()函数,主要是负责初始化计算函数并且重设其内部状态,一般就是重设其内部字段。一般在静态类中定义一个内部字段存放最终结果。
2.2:iterate()函数接收传入的参数,并进行内部轮转。其返回类型为boolean,每一次对一个新值进行聚集计算时候都会调用该方法,计算函数会根据聚集计算结果更新内部状态。当输入值合法或者正确计算了,则返回true。
2.3:terminatePartial()无参数,其为iterate()函数轮转结束后,返回轮转数据,terminatePartial()类似于Hadoop的Combiner,Hive需要部分结果的时候会调用该方法,必须要返回一个封装了聚集计算当前状态的对象。
2.4:merge接收terminatePartial()的返回结果,进行数据merge()操作,其返回类型为boolean,Hive进行合并一个部分聚集和另一个部分聚集的时候会调用该方法。
2.5:terminate()返回最终的聚集函数结果,Hive最终聚集结果的时候会调用该方法。计算函数需要把状态作为一个值返回给用户。

注:部分聚集结果的数据类型和最终结果的数据类型可以不同。

示例:自定义一个求最大值的UDAF函数:

/** 
 * UDAF是输入多个数据行,产生一个数据行 
 * 用户自定义UDAF必须继承UDFA,必须提供一个实现了UDAFEvaluator接口的内部类 
 * @author 廖*民 
 * time : 2015年1月30日下午9:43:38 
 * @version 
 */  
public class MaxiNumber extends UDAF{   
    public static class MaxiNumberIntUDAFEvaluator implements UDAFEvaluator{   
        //最终结果   
        private FloatWritable result;   
        //负责初始化计算函数并设置它的内部状态,result是存放最终结果的   
        public void init() {   
            result=null;   
        }   
        //每次对一个新值进行聚集计算都会调用iterate方法   
        public boolean iterate(FloatWritable value)   
        {   
            if(value==null)   
                return false;   
            if(result==null)   
              result=new FloatWritable(value.get());   
            else  
              result.set(Math.max(result.get(), value.get()));   
            return true;   
        }   
                                                                                                                                     
        //Hive需要部分聚集结果的时候会调用该方法   
        //会返回一个封装了聚集计算当前状态的对象   
        public FloatWritable terminatePartial()   
        {   
            return result;   
        }   
        //合并两个部分聚集值会调用这个方法   
        public boolean merge(FloatWritable other)   
        {   
            return iterate(other);   
        }   
        //Hive需要最终聚集结果时候会调用该方法   
        public FloatWritable terminate()   
        {   
            return result;   
        }   
    }   
}  

#打成jar包,上传到服务器。
#添加到Hive环境

hive (hive)> add jar /usr/local/src/maxi.jar;  
Added /usr/local/src/maxi.jar to class path  
Added resource: /usr/local/src/maxi.jar  

#创建临时函数

hive (hive)> create temporary function maxNumber as 'com.lixue.udaf.MaxiNumber';  
OK  
Time taken: 0.018 seconds  

#测试数据:求id的最大值

hive (hive)> select * from user;  
OK  
id  name  
1   lavimer  
2   liaozhongmin  
3   liaozemin  
100 hello  
200 hadoop  

#使用自定义的UDAF

hive (hive)> select maxNumber(id) from user;  
//MapReduce...  
OK  
_c0  
200.0  
Time taken: 5.364 seconds  

#删除UDAF

hive (hive)> drop temporary function maxNumber;  
OK  
Time taken: 0.03 seconds  
hive (hive)>  
上一篇下一篇

猜你喜欢

热点阅读