Hive自定义UDF函数

Posted 2023-05-13

参考技术A 在hive中，有时候一些内置的函数，和普通的查询操作已经满足不了我们要查询的要求，这时候可以自己写一些自定义函数来处理。自定义函数（user defined function =UDF）

由于hive本身是用java语言开发，所以udf必须用java来写才可以。

Hive中有三种UDF

编写UDF必须满足一下:

下面展示如何编写一个普通的UDF

1、编写类，继承UDF类，重写evaluate方法

2、 打包后上传到客户端所在的目录

3、使用该函数前，需要执行以下两行命令：

4、在sql中使用该函数

hive自定义函数UDF UDTF UDAF

Hive 自定义函数 UDF UDTF UDAF

1.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;

2.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;

3.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;