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;
}
}
}