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

1
 hive>add jar dop-udfdb.jar;

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

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

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
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;
    }
}