hive自定义函数将查询统计结果输出到指定的jdbc数据源

83 阅读 0 评论 55 点赞

我是靠谱客的博主沉默唇膏，最近开发中收集的这篇文章主要介绍hive自定义函数将查询统计结果输出到指定的jdbc数据源，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

如题：
1、自定义类继承抽象类GenericUDF
2、实现抽象类的三个方法
3、将其打成jar包（打成jar包时不需要指定main）eg: dop-udfdb.jar
4、进入hive命令行执行下面的命令将该ja包加入到hive环境

 hive>add jar dop-udfdb.jar;

5、添加jdbc驱动支持,这里mysql 驱动,不同的数据源添加各自的驱动包

hive>add jar mysql-connector-java-5.1.38.jar;

6、创建hive临时函数,指定函数名“udfdb”，处理函数的类’com.mayflay.AnalyzeGenericUDFDBOutput’,临时函数只在此会话有效，如果要长期有效删除关键字“temporary”

hive>create temporary function udfdb as 'com.mayflay.AnalyzeGenericUDFDBOutput';

7、使用自定义函数,分组统计，将统计的结果放到mysql库中，执行mysql的insert语句

hive>select udfdb('jdbc:mysql://localhost:3306/operator','root','root','insert into operator values(?,?,?)',account,tell,count(1)) from catering_call group by account,tell;

类实现

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.serde.Constants;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.StringObjectInspector;
import org.apache.hadoop.io.Text;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @Author: mihuajun 【kobe96688@126.com】
 * @Date: 11/21/2016 9:37 AM
 */
public class AnalyzeGenericUDFDBOutput extends GenericUDF {
    private transient ObjectInspector[] argumentOI;
    private transient Connection connection = null;
    private String url;
    private String user;
    private String pass;
    private Text result;


    @Override
    public ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentTypeException {
        argumentOI = arguments;
        //验证输入参数和该函数的返回类型
        return PrimitiveObjectInspectorFactory.writableStringObjectInspector;
    }


    @Override
    public Object evaluate(DeferredObject[] arguments) throws HiveException {
    //定义函数的输出过程
        url = ((StringObjectInspector) argumentOI[0]).getPrimitiveJavaObject(arguments[0].get());
        user = ((StringObjectInspector) argumentOI[1]).getPrimitiveJavaObject(arguments[1].get());
        pass = ((StringObjectInspector) argumentOI[2]).getPrimitiveJavaObject(arguments[2].get());
        try {
            connection = DriverManager.getConnection(url, user, pass);
        } catch (SQLException ex) {
            ex.printStackTrace();
        }
        if (connection != null) {
            try {
                PreparedStatement ps = connection.prepareStatement(((StringObjectInspector) argumentOI[3]).getPrimitiveJavaObject(arguments[3].get()));
                StringBuilder sb = new StringBuilder();
                for (int i = 4; i < arguments.length; ++i) {
                    PrimitiveObjectInspector poi = ((PrimitiveObjectInspector) argumentOI[i]);
                    ps.setObject(i - 3, poi.getPrimitiveJavaObject(arguments[i].get()));
                    sb.append(poi.getPrimitiveJavaObject(arguments[i].get()));
                    sb.append(" ");
                }
                result = new Text(sb.toString());
                ps.execute();
                ps.close();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    connection.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }

        return result;
    }

    @Override
    public String getDisplayString(String[] children) {
        //执行explain时会执行，和java中的tostring一样
        return null;
    }
}