天天看点
返回

[Hive]Hive自定义函数UDF

版权声明：本文为博主原创文章，未经博主允许不得转载。 https://blog.csdn.net/SunnyYoona/article/details/53244868

当Hive提供的内置函数无法满足你的业务处理需要时，此时就可以考虑使用用户自定义函数

用户自定义函数（user defined function)，针对单条记录。 

编写一个UDF，需要继承UDF类，并实现evaluate()函数。在查询执行过程中，查询中对应的每个应用到这个函数的地方都会对这个类进行实例化。对于每行输入都会调用到evaluate()函数。而evaluate()函数处理的值会返回给Hive。同时用户是可以重载evaluate方法的。Hive会像Java的方法重载一样，自动选择匹配的方法。

1. 简单UDF

1.1 自定义Java类

下面自定义一个Java类OperationAddUDF，实现了Int，Double，Float以及String类型的加法操作。

  1. package com.sjf.open.hive.udf;

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

  4. import org.apache.hadoop.hive.serde2.ByteStream;

  5. import org.apache.hadoop.hive.serde2.io.DoubleWritable;

  6. import org.apache.hadoop.hive.serde2.lazy.LazyInteger;

  7. import org.apache.hadoop.io.FloatWritable;

  8. import org.apache.hadoop.io.IntWritable;

  9. import org.apache.hadoop.io.Text;

  11. /**

  12. * Created by xiaosi on 16-11-19.

  13. */

  14. public class OperationAddUDF extends UDF {

  16.    private final ByteStream.Output out = new ByteStream.Output();

  17.    /**

  18.     * IntWritable

  19.     * @param num1

  20.     * @param num2

  21.     * @return

  22.     */

  23.    public IntWritable evaluate(IntWritable num1, IntWritable num2){

  24.        if(num1 == null || num2 == null){

  25.            return null;

  26.        }

  27.        return new IntWritable(num1.get() + num2.get());

  28.    }

  30.    /**

  31.     * DoubleWritable

  32.     * @param num1

  33.     * @param num2

  34.     * @return

  35.     */

  36.    public DoubleWritable evaluate(DoubleWritable num1, DoubleWritable num2){

  37.        if(num1 == null || num2 == null){

  38.            return null;

  39.        }

  40.        return new DoubleWritable(num1.get() + num2.get());

  41.    }

  43.    /**

  44.     * FloatWritable

  45.     * @param num1

  46.     * @param num2

  47.     * @return

  48.     */

  49.    public FloatWritable evaluate(FloatWritable num1, FloatWritable num2){

  50.        if(num1 == null || num2 == null){

  51.            return null;

  52.        }

  53.        return new FloatWritable(num1.get() + num2.get());

  54.    }

  56.    /**

  57.     * Text

  58.     * @param num1

  59.     * @param num2

  60.     * @return

  61.     */

  62.    public Text evaluate(Text num1, Text num2){

  63.        if(num1 == null || num2 == null){

  64.            return null;

  65.        }

  66.        try{

  67.            Integer n1 = Integer.valueOf(num1.toString());

  68.            Integer n2 = Integer.valueOf(num2.toString());

  69.            Integer result = n1 + n2;

  70.            out.reset();

  71.            LazyInteger.writeUTF8NoException(out, result);

  72.            Text text = new Text();

  73.            text.set(out.getData(), 0, out.getLength());

  74.            return text;

  75.        }

  76.        catch (Exception e){

  77.            return null;

  78.        }

  80.    }

  81. }

UDF中evaluate()函数的参数和返回值类型只能是Hive可以序列化的数据类型。例如，如果用户处理的全是数值，那么UDF的输出参数类型可以是基本数据类型int，Integer封装的对象或者是一个IntWritable对象，也就是Hadoop对整型封装后的对象。用户不需要特别的关心将调用到哪个类型，因为当类型不一致的时候，Hive会自动将数据类型转换成匹配的类型。null值在Hive中对于任何数据类型都是合法的，但是对于Java基本数据类型，不能是对象，也不能是null。

1.2 Hive中使用

如果想在Hive中使用UDF，那么需要将Java代码进行编译，然后将编译后的UDF二进制类文件打包成一个Jar文件。然后，在Hive会话中，将这个Jar文件加入到类路径下，在通过CREATE FUNCTION 语句定义好使用这个Java类的函数：

1.2.1 添加Jar文件到类路径下

  1. hive (test)> add jar /home/xiaosi/open-hive-1.0-SNAPSHOT.jar;

  2. Added [/home/xiaosi/open-hive-1.0-SNAPSHOT.jar] to class path

  3. Added resources: [/home/xiaosi/open-hive-1.0-SNAPSHOT.jar]

需要注意的是，Jar文件路径是不需要用引号括起来的，同时，到目前为止这个路径需要是当前文件系统的全路径。Hive不仅仅将这个Jar文件加入到classpath下，同时还将其加入到分布式缓存中，这样整个集群的机器都是可以获得该Jar文件的。

1.2.2 创建函数add

  1. hive (test)> create temporary function add as 'com.sjf.open.hive.udf.OperationAddUDF';

  2. OK

  3. Time taken: 0.004 seconds

注意的是create temporary function语句中的temporary关键字，当前会话中声明的函数只会在当前会话中有效。因此用户需要在每个会话中都增加Jar文件然后创建函数。不过如果用户需要频繁的使用同一个Jar文件和函数的话，那么可以将相关语句增加到$HOME/.hiverc文件中去。

1.2.3 使用

现在这个数值相加函数可以像其他的函数一样使用了。

  1. hive (test)> select add(12, 34) from employee_part;

  2. OK

  3. 46

  4. Time taken: 0.078 seconds, Fetched: 1 row(s)

  5. hive (test)> select add(12.3, 20.1) from employee_part;

  6. OK

  7. 32.400000000000006

  8. Time taken: 0.098 seconds, Fetched: 1 row(s)

  9. hive (test)> select add("12", "45") from employee_part;

  10. OK

  11. 57

  12. Time taken: 0.077 seconds, Fetched: 1 row(s)

1.2.4 删除UDF

当我们使用完自定义UDF后，我们可以通过如下命令删除此函数：

  1. hive (test)> drop temporary function if exists add;
2. 复杂UDF

2.1 GenericUDF

和UDF相比，GenericUDF（org.apache.hadoop.hive.ql.udf.generic.GenericUDF）支持复杂类型（比如List，struct，map等）的输入和输出。GenericUDF可以让我们通过ObjectInspector来管理方法的参数，检查接收参数的类型和数量。

GenericUDF要求实现一下三个方法：

  1. // this is like the evaluate method of the simple API. It takes the actual arguments and returns the result

  2. abstract Object evaluate(GenericUDF.DeferredObject[] arguments);

  4. // Doesn't really matter, we can return anything, but should be a string representation of the function.

  5. abstract String getDisplayString(String[] children);

  7. // called once, before any evaluate() calls. You receive an array of object inspectors that represent the arguments of the function

  8. // this is where you validate that the function is receiving the correct argument types, and the correct number of arguments.

  9. abstract ObjectInspector initialize(ObjectInspector[] arguments);

2.2 Example

我们想要在Hive实现一个strContain方法，需要两个参数，一个是包含字符串的列表（list<String>），另一个是待寻找的字符串（String）。如果列表中包含我们提供的字符串，返回tue，否则返回false。功能如下所示：

  1. strContain(List("a", "b", "c"), "b"); // true

  2. strContain(List("a", "b", "c"), "d"); // false

2.3 代码

  1. package com.sjf.open.hive.udf;

  3. import org.apache.hadoop.hive.ql.exec.Description;

  4. import org.apache.hadoop.hive.ql.exec.UDFArgumentException;

  5. import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;

  6. import org.apache.hadoop.hive.ql.metadata.HiveException;

  7. import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;

  8. import org.apache.hadoop.hive.serde2.lazy.LazyString;

  9. import org.apache.hadoop.hive.serde2.objectinspector.ListObjectInspector;

  10. import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;

  11. import org.apache.hadoop.hive.serde2.objectinspector.primitive.BooleanObjectInspector;

  12. import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

  13. import org.apache.hadoop.hive.serde2.objectinspector.primitive.StringObjectInspector;

  14. import org.apache.hadoop.io.BooleanWritable;

  16. import com.google.common.base.Objects;

  17. import org.slf4j.Logger;

  18. import org.slf4j.LoggerFactory;

  20. import java.util.List;

  22. /**

  23. * Created by xiaosi on 16-11-21.

  24. *

  25. */

  26. @Description(name = "contain", value = "_FUNC_(List<T>, T) ")

  27. public class GenericUDFStrContain extends GenericUDF {

  29. private static final Logger logger = LoggerFactory.getLogger(GenericUDFStrContain.class);

  31. private ListObjectInspector listObjectInspector;

  32. private StringObjectInspector stringObjectInspector;

  34. @Override

  35. public ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentException {

  37. logger.info("--------- GenericUDFStrContain --- initialize");

  39. // 参数个数校验

  40. if (arguments.length != 2) {

  41. throw new UDFArgumentLengthException(

  42. "The function 'Contain' only accepts 2 argument : List<T> and T , but got " + arguments.length);

  43. }

  45. ObjectInspector argumentOne = arguments[0];

  46. ObjectInspector argumentTwo = arguments[1];

  48. // 参数类型校验

  49. if (!(argumentOne instanceof ListObjectInspector)) {

  50. throw new UDFArgumentException("The first argument of function must be a list / array");

  51. }

  52. if (!(argumentTwo instanceof StringObjectInspector)) {

  53. throw new UDFArgumentException("The second argument of function must be a string");

  54. }

  56. this.listObjectInspector = (ListObjectInspector) argumentOne;

  57. this.stringObjectInspector = (StringObjectInspector) argumentTwo;

  59. // 链表元素类型检查

  60. if (!(listObjectInspector.getListElementObjectInspector() instanceof StringObjectInspector)) {

  61. throw new UDFArgumentException("The first argument must be a list of strings");

  62. }

  64. // 返回值类型

  65. return PrimitiveObjectInspectorFactory.javaBooleanObjectInspector;

  67. }

  69. @Override

  70. public Object evaluate(DeferredObject[] arguments) throws HiveException {

  72. logger.info("--------- GenericUDFStrContain --- evaluate");

  74. // 利用ObjectInspector从DeferredObject[]中获取元素值

  75. List<LazyString> list = (List<LazyString>) this.listObjectInspector.getList(arguments[0].get());

  76. String str = this.stringObjectInspector.getPrimitiveJavaObject(arguments[1].get());

  78. if (Objects.equal(list, null) || Objects.equal(str, null)) {

  79. return null;

  80. }

  82. // 判断是否包含查询元素

  83. for (LazyString lazyString : list) {

  84. String s = lazyString.toString();

  85. if (Objects.equal(str, s)) {

  86. return new Boolean(true);

  87. }

  88. }

  89. return new Boolean(false);

  91. }

  93. @Override

  94. public String getDisplayString(String[] children) {

  95. return "arrayContainsExample() strContain(List<T>, T)";

  96. }

  98. }

2.4 测试

Java测试：

  1. package com.sjf.open.hive.udf;

  4. import org.apache.hadoop.hive.ql.metadata.HiveException;

  5. import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;

  6. import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;

  7. import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;

  8. import org.apache.hadoop.hive.serde2.objectinspector.primitive.BooleanObjectInspector;

  9. import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

  11. import java.util.ArrayList;

  12. import java.util.List;

  14. /**

  15. * Created by xiaosi on 16-11-22.

  16. */

  17. public class GenericUDFStrContainTest {

  19. public static void test() throws HiveException {

  21. GenericUDFStrContain genericUDFStrContain = new GenericUDFStrContain();

  22. ObjectInspector stringOI = PrimitiveObjectInspectorFactory.javaStringObjectInspector;

  23. ObjectInspector listOI = ObjectInspectorFactory.getStandardListObjectInspector(stringOI);

  24. BooleanObjectInspector resultInspector = (BooleanObjectInspector) genericUDFStrContain.initialize(new ObjectInspector[]{listOI, stringOI});

  26. // create the actual UDF arguments

  27. List<String> list = new ArrayList<String>();

  28. list.add("a");

  29. list.add("b");

  30. list.add("c");

  32. // test our results

  34. // the value exists

  35. Object result = genericUDFStrContain.evaluate(new GenericUDF.DeferredObject[]{new GenericUDF.DeferredJavaObject(list), new GenericUDF.DeferredJavaObject("a")});

  36. System.out.println("-----------" + result);

  38. // the value doesn't exist

  39. Object result2 = genericUDFStrContain.evaluate(new GenericUDF.DeferredObject[]{new GenericUDF.DeferredJavaObject(list), new GenericUDF.DeferredJavaObject("d")});

  40. System.out.println("-----------" + result2);

  42. // arguments are null

  43. Object result3 = genericUDFStrContain.evaluate(new GenericUDF.DeferredObject[]{new GenericUDF.DeferredJavaObject(null), new GenericUDF.DeferredJavaObject(null)});

  44. System.out.println("-----------" + result3);

  45. }

  47. public static void main(String[] args) throws HiveException {

  48. test();

  49. }

  50. }

Hive测试：

在Hive中使用跟简单UDF一样，需要将Java代码进行编译，然后将编译后的UDF二进制类文件打包成一个Jar文件。然后，在Hive会话中，将这个Jar文件加入到类路径下，在通过CREATE FUNCTION 语句定义好使用这个Java类的函数：

  1. hive (test)> add jar /home/xiaosi/code/openDiary/HiveCode/target/open-hive-1.0-SNAPSHOT.jar;

  2. Added [/home/xiaosi/code/openDiary/HiveCode/target/open-hive-1.0-SNAPSHOT.jar] to class path

  3. Added resources: [/home/xiaosi/code/openDiary/HiveCode/target/open-hive-1.0-SNAPSHOT.jar]

  4. hive (test)> create temporary function strContain as 'com.sjf.open.hive.udf.GenericUDFStrContain';

  5. OK

  6. Time taken: 0.021 seconds

使用：

  1. hive (test)> select subordinates, strContain(subordinates, "tom") from employee2;

  2. OK

  3. ["lily","lucy","tom"] true

  4. ["lucy2","tom2"] false

  5. ["lily","lucy","tom"] true

  6. ["lily","yoona","lucy"] false

  7. Time taken: 1.147 seconds, Fetched: 4 row(s)

  8. hive (test)> select subordinates, strContain(subordinates, 1) from employee2;

  9. FAILED: SemanticException [Error 10014]: Line 1:21 Wrong arguments '1': The second argument of function must be a string

  10. hive (test)> select subordinates, strContain("yoona", 1) from employee2;

  11. FAILED: SemanticException [Error 10014]: Line 1:21 Wrong arguments '1': The first argument of function must be a list / array

  12. hive (test)> select subordinates, strContain("yoona", 1, 3) from employee2;

  13. FAILED: SemanticException [Error 10015]: Line 1:21 Arguments length mismatch '3': The function 'Contain' only accepts 2 argument : List<T> and T , but got 3

备注：

subordinates是一个array<string>类型集合。

资料：

http://blog.matthewrathbone.com/2013/08/10/guide-to-writing-hive-udfs.html#the-complex-api
sql java hive 缓存 分布式计算 hadoop udf hive函数 udf定义函数 udf hive自定义函数 hadoop java图文详解 hive udf权限问题
上一篇: mybaits源码分析--binding模块（五）
下一篇: [Hadoop]输入路径过滤，通配符与PathFilter

继续阅读