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

偏执的太偏执、 2022-07-14 01:25 139阅读 0赞

如题：  
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;
        }
    }