最近组内童鞋在自定义的Hive的UDF、UDTF，然后Spark SQL去使用时候，遇到种种问题。躺了一遍血坑之后后来发现UDTF是一种机制，UDF、UDAF是一种机制。绝对一部血泪史，非常值得记录和深入探究。于是计划带着问题研读下Hive UDF管理机制，以及如何Spark SQL和Hive的UDF结合。

大部分时候我们在写SQL时候是直接调用框架已有的函数，例如sum()，count()，但是有写业务场合，我们希望有其它更多的函数。因为Hive执行QL语句就可以执行MapReduce，我们自然希望SQL语句的功能更加强大，其结果就是我们想要的结果，这时，我们就需要自定义一些函数了，Hive是对外开放了三种自定义函数接口UDF、UDAF、UDTF。

一：什么是UDF、UDAF、UDTF？

1，UDF：User-Defined Function

特点：一进一出；

继承UDF类（org.apache.hadoop.hive.ql.exec.UDF）

2，UDAF：User- Defined Aggregation Funcations

特点：多进一出

继承UDAF类（org.apache.hadoop.hive.ql.exec.UDAF）

3，UDTF：User-Defined Table-Generating Functions

特点：一进多出

继承UDTF类（ org.apache.hadoop.hive.ql.udf.generic.GenericUDTF）

通过继承关系能看出UDF和UDAF是一个系列，UDTF则是另外一种系列。

二：UDF

如何开发一个简单UDF

自定义UDF需要继承org.apache.hadoop.hive.ql.UDF。
需要实现evaluate函。
evaluate函数支持重载。

做一个简单字符串长度的DEMO如下：

import org.apache.hadoop.hive.ql.exec.UDF;

public final class GetStrLen extends UDF{
    public Integer evaluate(Sting str) {
        if(str!=null){
            return str.length();
        }else{
            return ;
            }
    }
}

org.apache.hadoop.hive.ql.exec.UDF源码如下：

UDF源码中有两个构造器，分别是有参和无参的。而有参数的参数是UDFMethodResolver。那么什么是UDFMethodResolver 呢？看javadoc描述：

/**
* The UDF Method resolver interface. A user can plugin a resolver to their UDF
* by implementing the functions in this interface. Note that the resolver is
* stored in the UDF class as an instance variable. We did not use a static
* variable because many resolvers maintain the class of the enclosing UDF as
* state and are called from a base class e.g. UDFBaseCompare. This makes it
* very easy to write UDFs that want to do resolution similar to the comparison
* operators. Such UDFs just need to extend UDFBaseCompare and do not have to
* care about the UDFMethodResolver interface. Same is true for UDFs that want
* to do resolution similar to that done by the numeric operators. Such UDFs
* simply have to extend UDFBaseNumericOp class. For the default resolution the
* UDF implementation simply needs to extend the UDF class.
*/

UDFMethodResolver是UDF类的解析接口可以自定义输入数据的类型，如果UDF的构造中没有传入实现了UDFMethodResolver接口的类的话，会指定一个默认的DefaultUDFMethodResolver，当然，也可以通过使用setResolver(UDFMethodResolver rslv)函数指定一个自定义的UDFMethodResolver，这个函数也是可覆写的。

三：UDAF

1，利用抽象类UDAF和UDAFEvaluator方式开发一个简单的UDAF

定义静态内部类实现UDAFEvaluator的方法如下步骤：

init() 初始化一般负责初始化内部字段,通常初始化用来存放最终结果的变量
iterate() 每次都会对一个新的值进行聚合计算时都调用该方法,一般会根据计算结果更新用来存放最终结果的变量,如果计算正确或者输入值合法就返回true
terminatePartial() 这个方法直译过来是"终止部分",部分聚合结果的时候调用该方法必须返回一个封装了聚合计算当前状态的对象,类似于 MapReduce的combiner
merge() 接受来自 terminatePartial的返回结果,进行合并,hive合并两部分聚合的时候回调用这个方法
terminate() 终止方法 返回最终聚合函数结果

简单的求平均值例子

import org.apache.hadoop.hive.ql.exec.UDAFEvaluator;
import org.apache.hadoop.hive.ql.exec.UDAF;
public class getAvg extends UDAF{
    //数据类型可以是ArrayList<String> and HashMap<String,Double>
    public static class UDAFAvgState{
      private long mCount;
      private double mSum;
  }
  //实现聚合操作的真正的方法。hive自动查找实现UDAFEvaluate的udaf内部类
  public static class UDAFGetAvg implements UDAFEvaluate{
    UDAFAvgState state;
    public UDAFGetAvg(){
      super();
      state = new UDAFAvgState();
      init();
  }
   //初始化方法
   public void init(){
     state.mSum = ；
     state.mCount = ; 
  }
  //循环访问一行原始数据
  //参数类型和数目和udaf函数保持一致
  //该方法的返回值肯定是true
  public boolean iterate(Double o){
    if (o != null){
      state.mSum +=o;
      state.mCount++;
   }
   return true;
 }
 //返回部分聚合的结果,若果是primitive类型，返回java类，如Integer或String
 public UDAFAvgState terminatePartial(){
 //0的平均值为空
  return state.mCount == ? null : state;

}
//部分聚合合并
//参数值为terminatePartial方法的返回值
public boolean merge(UDAFAvgState o){
if (o != null){
  state.mSum +=o.mSum;
  state.mCount +=o.mCount;
 }
 return true;

}
//终止聚合,返回聚合结果
public Double terminate(){
    return state.mCount == ? null : Double.valueOf(state.mSum / state.mCount);
   } 
 } 
}

那么其实还有第二种形式自定义UDAF利用接口GenericUDAFResolver2（或者抽象类AbstractGenericUDAFResolver）和抽象类GenericUDAFEvaluator。

2，基于GenericUDAFResolver2开发简单UDAF

第一个是编写resolver类，第二个是编写evaluator类。

resolver负责类型检查，操作符重载。evaluator真正实现UDAF的逻辑。通常来说，顶层UDAF类继承org.apache.hadoop.hive.ql.udf.GenericUDAFResolver2，里面编写嵌套类evaluator 实现UDAF的逻辑。

public class GenericUDAFSum extends AbstractGenericUDAFResolver {  
  
  static final Log LOG = LogFactory.getLog(GenericUDAFSum.class.getName());  
  
  @Override  
  public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters)  
    throws SemanticException {  
    // Type-checking goes here!  
    return new GenericUDAFSumLong();   
  }   
  
  public static class GenericUDAFSumLong extends GenericUDAFEvaluator {  
    // UDAF logic goes here!  
  }   
}

比如说，自己写了一个SUM，SUM对于Long类型和Double类型进行求和，没有问题。但是，如果传入的参数是一个Array呢？这个时候，就需要在Evaluator方法里面，对参数进行校验了。

public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters)  
      throws SemanticException {  
    if (parameters.length != 1) {  
      throw new UDFArgumentTypeException(parameters.length - 1,  
          "Exactly one argument is expected.");  
    }  
  
    if (parameters[].getCategory() != ObjectInspector.Category.PRIMITIVE) {  
      throw new UDFArgumentTypeException(,  
          "Only primitive type arguments are accepted but "  
          + parameters[].getTypeName() + " is passed.");  
    }  
    switch (((PrimitiveTypeInfo) parameters[]).getPrimitiveCategory()) {  
    case BYTE:  
    case SHORT:  
    case INT:  
    case LONG:  
    case FLOAT:  
    case DOUBLE:  
    case STRING:  
    case TIMESTAMP:  
      return new GenericUDAFAverageEvaluator();  
    case BOOLEAN:  
    default:  
      throw new UDFArgumentTypeException(,  
          "Only numeric or string type arguments are accepted but "  
          + parameters[].getTypeName() + " is passed.");  
    }  
  }

这个方法只支持Primitive类型，也就是INT，String，Double，Float这些。

UDAF使用一个ObjectInspector来抽象化每一行数据的读取。上面使用的Primitive类型的数据，所以使用PrimitiveObjectInspector来读取传入的参数。

UDAF会根据不同的计算模型，产生不同的阶段。

如：SUM()聚合函数，接受一个原始类型的整型数值，然后创建一个整型的PARTIAL数据，
返回一个固定的整型结果。

如：median() 中位数
可以接受原始整型输入，然后会产生一个中间的整数PARTIAL数据(排序)，
然后再返回一个固定的整型结果。

UDAF的Mode，也就是执行阶段。无论怎样的UDAF，最终都会变成MapReduce Job。Mode是一UDAF的使用类型，主要有4种形势：因为MapReduce可能是，Map->Reduce也可能是，Map->Reduce->Reduce

public static enum Mode {  
    /** 
     * PARTIAL1: 这个是mapreduce的map阶段:从原始数据到部分数据聚合 
     * 将会调用iterate()和terminatePartial() 
     */  
    PARTIAL1,  
        /** 
     * PARTIAL2: 这个是mapreduce的map端的Combiner阶段，负责在map端合并map的数据::从部分数据聚合到部分数据聚合: 
     * 将会调用merge() 和 terminatePartial()  
     */  
    PARTIAL2,  
        /** 
     * FINAL: mapreduce的reduce阶段:从部分数据的聚合到完全聚合  
     * 将会调用merge()和terminate() 
     */  
    FINAL,  
        /** 
     * COMPLETE: 如果出现了这个阶段，表示mapreduce只有map，没有reduce，所以map端就直接出结果了:从原始数据直接到完全聚合 
      * 将会调用 iterate()和terminate() 
     */  
    COMPLETE  
};

有的UDAF函数会可以像UDF函数那样使用，有的必须在聚合函数环境下使用，如group by，over(partition by )

而在使用UDAF进行计算的时候，会启用一个init方法。这个init的方法会在买个阶段前面都启动一次。第一次启动的时候，参数指的是读入每一行记录的参数。第二次启动的时候，传入的参数只有1个，指的是中间结果的参数。这里需要特别注意。

@Override  
        public ObjectInspector init(Mode m, ObjectInspector[] parameters)  
                throws HiveException {  
            super.init(m, parameters);  
              
            //init input  
            if (m == Mode.PARTIAL1 || m == Mode.COMPLETE){ //必须得有  
                LOG.info(" Mode:"+m.toString()+" result has init");  
                inputOI = (PrimitiveObjectInspector) parameters[];  
                inputOI2 = (PrimitiveObjectInspector) parameters[1];  
//              result = new DoubleWritable(0);  
            }  
            //init output  
            if (m == Mode.PARTIAL2 || m == Mode.FINAL) {  
                outputOI = (PrimitiveObjectInspector) parameters[];  
                result = new DoubleWritable();  
                return PrimitiveObjectInspectorFactory.writableDoubleObjectInspector;  
            }else{  
                result = new DoubleWritable();  
                return PrimitiveObjectInspectorFactory.writableDoubleObjectInspector;  
            }   
              
        }

最后是完整的UDAF代码。实现一个有条件的SUM，传入2个参数，当第二个参数>1 的时候进行SUM。

import org.apache.commons.logging.Log;  
import org.apache.commons.logging.LogFactory;  
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;  
import org.apache.hadoop.hive.ql.metadata.HiveException;  
import org.apache.hadoop.hive.ql.parse.SemanticException;  
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;  
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFParameterInfo;  
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFResolver2;  
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;  
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils;  
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;  
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorUtils.ObjectInspectorCopyOption;  
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;  
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorUtils;  
import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;  
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;  
import org.apache.hadoop.hive.serde2.io.DoubleWritable;  
import org.apache.hadoop.util.StringUtils;  
  
public class GenericUdafMemberLevel2 extends AbstractGenericUDAFResolver {  
    private static final Log LOG = LogFactory  
            .getLog(GenericUdafMemberLevel2.class.getName());  
      
    @Override  
      public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters)  
        throws SemanticException {  
          
        return new GenericUdafMeberLevelEvaluator();  
      }  
      
    public static class GenericUdafMeberLevelEvaluator extends GenericUDAFEvaluator {  
        private PrimitiveObjectInspector inputOI;  
        private PrimitiveObjectInspector inputOI2;  
        private PrimitiveObjectInspector outputOI;  
        private DoubleWritable result;  
  
        @Override  
        public ObjectInspector init(Mode m, ObjectInspector[] parameters)  
                throws HiveException {  
            super.init(m, parameters);  
              
            //init input  
            if (m == Mode.PARTIAL1 || m == Mode.COMPLETE){ //必须得有  
                LOG.info(" Mode:"+m.toString()+" result has init");  
                inputOI = (PrimitiveObjectInspector) parameters[];  
                inputOI2 = (PrimitiveObjectInspector) parameters[1];  
//              result = new DoubleWritable(0);  
            }  
            //init output  
            if (m == Mode.PARTIAL2 || m == Mode.FINAL) {  
                outputOI = (PrimitiveObjectInspector) parameters[];  
                result = new DoubleWritable();  
                return PrimitiveObjectInspectorFactory.writableDoubleObjectInspector;  
            }else{  
                result = new DoubleWritable();  
                return PrimitiveObjectInspectorFactory.writableDoubleObjectInspector;  
            }   
              
        }  
  
        /** class for storing count value. */  
        static class SumAgg implements AggregationBuffer {  
            boolean empty;  
            double value;  
        }  
  
        @Override  
        //创建新的聚合计算的需要的内存，用来存储mapper,combiner,reducer运算过程中的相加总和。  
        //使用buffer对象前，先进行内存的清空——reset  
        public AggregationBuffer getNewAggregationBuffer() throws HiveException {  
            SumAgg buffer = new SumAgg();  
            reset(buffer);  
            return buffer;  
        }  
  
        @Override  
        //重置为0  
        //mapreduce支持mapper和reducer的重用，所以为了兼容，也需要做内存的重用。  
        public void reset(AggregationBuffer agg) throws HiveException {  
            ((SumAgg) agg).value = .;  
            ((SumAgg) agg).empty = true;  
        }  
  
        private boolean warned = false;  
        //迭代  
        //只要把保存当前和的对象agg，再加上输入的参数，就可以了。  
        @Override  
        public void iterate(AggregationBuffer agg, Object[] parameters)  
                throws HiveException {  
            // parameters == null means the input table/split is empty  
            if (parameters == null) {  
                return;  
            }  
            try {  
                double flag = PrimitiveObjectInspectorUtils.getDouble(parameters[1], inputOI2);  
                if(flag > 1.)   //参数条件  
                    merge(agg, parameters[]);   //这里将迭代数据放入combiner进行合并  
              } catch (NumberFormatException e) {  
                if (!warned) {  
                  warned = true;  
                  LOG.warn(getClass().getSimpleName() + " "  
                      + StringUtils.stringifyException(e));  
                }  
              }  
  
        }  
  
        @Override  
        //这里的操作就是具体的聚合操作。  
        public void merge(AggregationBuffer agg, Object partial) {  
            if (partial != null) {  
                // 通过ObejctInspector取每一个字段的数据  
                if (inputOI != null) {  
                    double p = PrimitiveObjectInspectorUtils.getDouble(partial,  
                            inputOI);  
                    LOG.info("add up 1:" + p);  
                    ((SumAgg) agg).value += p;  
                } else {  
                    double p = PrimitiveObjectInspectorUtils.getDouble(partial,  
                            outputOI);  
                    LOG.info("add up 2:" + p);  
                    ((SumAgg) agg).value += p;  
                }  
            }  
        }  
  
  
        @Override  
        public Object terminatePartial(AggregationBuffer agg) {  
                return terminate(agg);  
        }  
          
        @Override  
        public Object terminate(AggregationBuffer agg){  
            SumAgg myagg = (SumAgg) agg;  
            result.set(myagg.value);  
            return result;  
        }  
    }  
}

ITPub博客

Hive UDF - 管理机制

一：什么是UDF、UDAF、UDTF？

二：UDF

三：UDAF

1，利用抽象类UDAF和UDAFEvaluator方式开发一个简单的UDAF

2，基于GenericUDAFResolver2开发简单UDAF

最新文章