目录

Hive

简介

  • 1）Hive 自带了一些函数，比如：max/min等，但是数量有限，自己可以通过自定义UDF来方便的扩展。
  • 2）当Hive提供的内置函数无法满足你的业务处理需要时，此时就可以考虑使用用户自定义函数（UDF：user-defined function）。
  • 3）根据用户自定义函数类别分为以下三种：
    • （1）UDF（User-Defined-Function）一进一出
    • （2）UDAF（User-Defined Aggregation Function）聚集函数，多进一出，类似于：count、max、min
    • （3）UDTF（User-Defined Table-Generating Functions）一进多出，如lateral view explode()
  • 4）官方文档地址：https://cwiki.apache.org/confluence/display/Hive/HivePlugins

编程步骤

  • （1）继承Hive提供的类
    • org.apache.hadoop.hive.ql.udf.generic.GenericUDF  
    • org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
  • （2）实现类中的抽象方法
  • （3）在hive的命令行窗口创建函数
    • 添加jar
    • add jar linux_jar_path
    • 创建function
    • create [temporary] function [dbname.]function_name AS class_name;
  • （4）在hive的命令行窗口删除函数
    • drop [temporary] function [if exists] [dbname.]function_name;

自定义UDF函数

0）需求:

自定义一个UDF实现计算给定字符串的长度，例如：


hive(default)> select my_len("abcd");
4

1）创建一个Maven工程Hive

2）导入依赖


<dependencies>
    <dependency>
        <groupId>org.apache.hive</groupId>
        <artifactId>hive-exec</artifactId>
        <version>3.1.2</version>
    </dependency>
</dependencies>

3）创建一个类


package com.cosyblogs.udf;


import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

/**
 * 插件性质的开发：
 * 1. 实现接口或者继承类
 * 2. 重写相应的方法
 * 3. 打包
 * <p>
 * <p>
 * 自定义UDF函数类
 * 继承Hive提供的GenericUDF类
 * <p>
 * <p>
 * mylen('abcd') -> 4
 */
public class CalStringLengthUDF extends GenericUDF {
    /**
     * 初始化方法
     *
     * @param arguments 传入到函数中的参数对应的类型的鉴别器对象
     * @return 指定函数的返回值类型对应的鉴别器对象
     * @throws UDFArgumentException
     */
    @Override
    public ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentException {
        //1. 校验函数的参数个数
        if (arguments == null || arguments.length != 1) {
            throw new UDFArgumentLengthException("函数的参数个数不正确!!!");
        }
        //2. 校验函数的参数类型
        if (!arguments[0].getCategory().equals(ObjectInspector.Category.PRIMITIVE)) {
            throw new UDFArgumentTypeException(0, "函数的参数类型不正确!!!");
        }
        //3. 返回函数的返回值类型对应的鉴别器类型
        return PrimitiveObjectInspectorFactory.javaIntObjectInspector;
    }


    /**
     * 函数核心处理方法
     *
     * @param arguments 传入到函数的参数
     * @return 函数的返回值
     * @throws HiveException
     */
    @Override
    public Object evaluate(DeferredObject[] arguments) throws HiveException {
        //1. 获取参数
        Object argument = arguments[0].get();
        if (argument == null) {
            return 0;
        }
        return argument.toString().length();
    }


    @Override
    public String getDisplayString(String[] children) {
        return "";
    }
}

4）打成jar包上传到服务器 /home/hadoop/hive/datas/hive-1.0-SNAPSHOT.jar

5）将jar包添加到hive的classpath


hive (default)> add jar /home/hadoop/jar/hive-1.0-SNAPSHOT.jar;

6）创建临时函数与开发好的java class关联


hive (default)> create temporary function my_len as "com.cosyblogs.udf.CalStringLengthUDF";

7）即可在hive中使用自定义的函数 


hive (default)> select ename,my_len(ename) ename_len from emp;
Hive 自定义函数-最码农

自定义UDTF函数

0）需求 

自定义一个UDTF实现将一个任意分割符的字符串切割成独立的单词，例如：


hive(default)> select myudtf("hello,world,hadoop,hive", ",");
hello
world
hadoop
hive

1）代码实现


package com.cosyblogs.udtf;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructField;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

import java.util.ArrayList;
import java.util.List;

/**
 * 自定义UDTF
 * 继承Hive提供的GenericUDTF类
 * <p>
 * select myudtf("hello,world,hadoop,hive", ",");
 */
public class SplitStringToRowsUDTF extends GenericUDTF {

    private List<String> outs = new ArrayList<String>();

    /**
     * 初始化方法
     *
     * @param argOIs
     * @return
     * @throws UDFArgumentException
     */
    @Override
    public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {
        //1.判断参数的个数
        List<? extends StructField> structFieldRefs = argOIs.getAllStructFieldRefs();
        if (structFieldRefs.size() != 2) {
            throw new UDFArgumentLengthException("函数的参数个数不正确!!!");
        }
        //2.判断参数的类型
        for (int i = 0; i < structFieldRefs.size(); i++) {
            StructField structFieldRef = structFieldRefs.get(i);
            if (!structFieldRef.getFieldObjectInspector().getCategory().equals(ObjectInspector.Category.PRIMITIVE)) {
                throw new UDFArgumentTypeException(i, "函数的参数类型不正确!!!");
            }
        }

        //3.
        // 用于指定列的名字
        List<String> structFieldNames = new ArrayList<String>();
        structFieldNames.add("word");
        // 用于指定列的类型
        List<ObjectInspector> structFieldObjectInspectors = new ArrayList<ObjectInspector>();
        structFieldObjectInspectors.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);

        return ObjectInspectorFactory.getStandardStructObjectInspector(structFieldNames, structFieldObjectInspectors);

    }

    /**
     * 函数的核心处理方法
     *
     * @param args 传入到函数中的参数
     * @throws HiveException
     */
    @Override
    public void process(Object[] args) throws HiveException {
        //1.获取第一参数
        String words = args[0].toString();   //"hello,world,hadoop,hive"
        //2.获取第二参数
        String split = args[1].toString();  // ","
        //3.切割
        String[] splitwords = words.split(split);  //[hello,world,hadoop,hive]
        //4.将每个单词作为一行数据写出
        for (String word : splitwords) {
            outs.clear();
            outs.add(word);
            forward(outs);
        }

    }

    @Override
    public void close() throws HiveException {

    }
}

4）打成jar包上传到服务器 /home/hadoop/hive/datas/hive-1.0-SNAPSHOT.jar

5）将jar包添加到hive的classpath


hive (default)> add jar /home/hadoop/jar/hive-1.0-SNAPSHOT.jar;

6）创建临时函数与开发好的java class关联


hive (default)> create temporary function my_split as "com.cosyblogs.udtf.SplitStringToRowsUDTF";

7）即可在hive中使用自定义的函数 


hive (default)> select my_split("hello,world,hadoop,hive",",");
Hive 自定义函数-最码农