Hive 自定义函数 UDF UDTF UDAF

悠悠 2024-04-18 13:20 26阅读 0赞

### Hive 自定义函数 UDF UDTF UDAF ###

*  UDF：用户定义（普通）函数，只对单行数值产生作用；
    
    UDF只能实现一进一出的操作。
    
        定义udf 计算两个数最小值
        
        public class Min extends UDF {
        
                public Double evaluate(Double a, Double b) {
        
                    if (a == null)
                        a = 0.0;
                    if (b == null)
                        b = 0.0;
                    if (a >= b) {
                        return b;
                    } else {
                        return a;
                    }
                }
            }
            
         
        a）把程序打成jar包
        
        b）添加jar包：add jar /run/jar/udf_test.jar;
        
        c）创建临时函数：hive>CREATE TEMPORARY FUNCTION add_example AS 'hive.udf.Add';
        
        d）销毁临时函数：hive> DROP TEMPORARY FUNCTION add_example;
 *  UDTF：User-Defined Table-Generating Functions，用户定义表生成函数
    
    用来解决输入一行输出多行
    
        继承org.apache.hadoop.hive.ql.udf.generic.GenericUDTF,
        实现initialize, process, close三个方法。
        
        UDTF首先会调用initialize方法，此方法返回UDTF的返回行的信息（返回个数，类型）。
        
        初始化完成后，会调用process方法,真正的处理过程在process函数中，在process中，每一次forward()调用产生一行；如果产生多列可以将多个列的值放在一个数组中，然后将该数组传入到forward()函数。
        
        最后close()方法调用，对需要清理的方法进行清理
    
        例子：切分”key:value;key:value”字符串，返回结果为key, value两个字段。
        import java.util.ArrayList;
        
         import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
         import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
         import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
         import org.apache.hadoop.hive.ql.metadata.HiveException;
         import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
         import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
         import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
         import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
        
         public class ExplodeMap extends GenericUDTF{
        
             @Override
             public void close() throws HiveException {
                 // TODO Auto-generated method stub    
             }
        
             @Override
             public StructObjectInspector initialize(ObjectInspector[] args)
                     throws UDFArgumentException {
                 if (args.length != 1) {
                     throw new UDFArgumentLengthException("ExplodeMap takes only one argument");
                 }
                 if (args[0].getCategory() != ObjectInspector.Category.PRIMITIVE) {
                     throw new UDFArgumentException("ExplodeMap takes string as a parameter");
                 }
        
                 ArrayList<String> fieldNames = new ArrayList<String>();
                 ArrayList<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>();
                 fieldNames.add("col1");
                 fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
                 fieldNames.add("col2");
                 fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        
                 return ObjectInspectorFactory.getStandardStructObjectInspector(fieldNames,fieldOIs);
             }
        
             @Override
             public void process(Object[] args) throws HiveException {
                 String input = args[0].toString();
                 String[] test = input.split(";");
                 for(int i=0; i<test.length; i++) {
                     try {
                         String[] result = test[i].split(":");
                         forward(result);
                     } catch (Exception e) {
                         continue;
                     }
                 }
             }
         }
    
        a）把程序打成jar包
        
        b）添加jar包：add jar /run/jar/udf_test.jar;
        
        c）创建临时函数：CREATE TEMPORARY FUNCTION explode_map AS 'cn.itcast.hive.udtf.ExplodeMap';
        
        d）销毁临时函数：hive> DROP TEMPORARY FUNCTION add_example;
    
    udtf的使用：
    
        UDTF有两种使用方法，一种直接放到select后面，一种和lateral view一起使用。
        
        create table src(properties String);
        	
        	vi src.txt
        	key1:value1;key2:value2;
        	
        load data local inpath '/root/hivedata/src.txt' into table src;
        
        1：直接select中使用
        
        select explode_map(properties) as (col1,col2) from src;
        
        不可以添加其他字段使用
        select a, explode_map(properties) as (col1,col2) from src;
        
        不可以嵌套调用
        select explode_map(explode_map(properties)) from src;
        
        不可以和group by/cluster by/distribute by/sort by一起使用
        select explode_map(properties) as (col1,col2) from src group by col1, col2;
         
        
        
        2：和lateral view一起使用
        
        select src.id, mytable.col1, mytable.col2 from src lateral view explode_map(properties) mytable as col1, col2;
 *  UDAF：User- Defined Aggregation Funcation；用户定义聚合函数，可对多行数据产生作用；
    
    等同与SQL中常用的SUM()，AVG()，也是聚合函数；
    
    UDAF实现多进一出
    
    UDAF实现有简单与通用两种方式：  
    import org.apache.hadoop.hive.ql.exec.UDAF;  
    import org.apache.hadoop.hive.ql.exec.UDAFEvaluator;  
    import org.apache.hadoop.io.IntWritable;

//UDAF是输入多个数据行，产生一个数据行  
//用户自定义的UDAF必须是继承了UDAF，且内部包含多个实现了exec的静态类  
public class MaxiNumber extends UDAF \{  
public static class MaxiNumberIntUDAFEvaluator implements UDAFEvaluator \{  
// 最终结果  
private IntWritable result;

// 负责初始化计算函数并设置它的内部状态，result是存放最终结果的
        @Override
        public void init() {
            result = null;
        }
    
        // 每次对一个新值进行聚集计算都会调用iterate方法
        public boolean iterate(IntWritable value) {
            if (value == null)
                return false;
            if (result == null)
                result = new IntWritable(value.get());
            else
                result.set(Math.max(result.get(), value.get()));
            return true;
        }
    
        // Hive需要部分聚集结果的时候会调用该方法
        // 会返回一个封装了聚集计算当前状态的对象
        public IntWritable terminatePartial() {
            return result;
        }
    
        // 合并两个部分聚集值会调用这个方法
        public boolean merge(IntWritable other) {
            return iterate(other);
        }
    
        // Hive需要最终聚集结果时候会调用该方法
        public IntWritable terminate() {
            return result;
        }
    }