使用hive向mysql，sqlserver，hbase插入数据

opo

package com.hive.udf.test;　　

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

　　
import org.apache.commons.logging.Log;
　　
import org.apache.commons.logging.LogFactory;
　　
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.serdeConstants;
　　
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.IntWritable;
　　

　　
public class GenericSQLServerDBOutput extends GenericUDF {
　　
    private static final Log LOG = LogFactory.getLog(
　　
                                    GenericSQLServerDBOutput.class.getName());
　　

　　
    private transient ObjectInspector[] argumentOI;//参数
　　
    private transient Connection connection = null;
　　
    private String url;//jdbc url
　　
    private String user;//用户名
　　
    private String pass;//密码
　　
    private String dbName;//数据库名称
　　
    private final IntWritable result = new IntWritable(-1);//返回参数
　　

　　
    //初始化方法，确定返回类型
　　
    @Override
　　
    public ObjectInspector initialize(ObjectInspector[] arguments)
　　
            throws UDFArgumentTypeException {
　　
        argumentOI = arguments;
　　

　　
        for (int i = 0; i < 4; i++) {
　　
            if (arguments.getCategory() == ObjectInspector.Category.PRIMITIVE) {
　　
                PrimitiveObjectInspector poi = ((PrimitiveObjectInspector) arguments);
　　

　　
                if (!(poi.getPrimitiveCategory() ==
　　
                            PrimitiveObjectInspector.PrimitiveCategory.STRING)) {
　　
                    throw new UDFArgumentTypeException(i,
　　
                            "The argument of function  should be \""
　　
                                    + serdeConstants.STRING_TYPE_NAME
　　
                                    + "\", but \"" + arguments.getTypeName()
　　
                                    + "\" is found");
　　
                }
　　
            }
　　
        }
　　

　　
        for (int i = 4; i < arguments.length; i++) {
　　
            if (arguments.getCategory() != ObjectInspector.Category.PRIMITIVE) {
　　
                throw new UDFArgumentTypeException(i,
　　
                        "The argument of function should be primative"
　　
                                + ", but \"" + arguments.getTypeName()
　　
                                + "\" is found");
　　
            }
　　
        }
　　

　　
        return PrimitiveObjectInspectorFactory.writableIntObjectInspector;
　　
    }
　　

　　
    @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());
　　
        dbName = ((StringObjectInspector) argumentOI[3])
　　
                .getPrimitiveJavaObject(arguments[3].get());
　　

　　
        try {
　　
            Class.forName("com.microsoft.sqlserver.jdbc.SQLServerDriver");
　　
            connection = DriverManager.getConnection(url + ";DatabaseName=" + dbName,
　　
                            user, pass);
　　
        } catch (SQLException ex) {
　　
            LOG.error("Driver loading or connection issue", ex);
　　
            result.set(2);
　　
        } catch (ClassNotFoundException e) {
　　
            e.printStackTrace();
　　
        }
　　

　　
        if (connection != null) {
　　
            try {
　　
                PreparedStatement ps =
　　
                    connection.prepareStatement(((StringObjectInspector) argumentOI[4])
　　
                                        .getPrimitiveJavaObject(arguments[4].get()));
　　

　　
                for (int i = 5; i < arguments.length; ++i) {
　　
                    PrimitiveObjectInspector poi =
　　
                        ((PrimitiveObjectInspector) argumentOI);
　　
                      ps.setObject(i - 4, poi.getPrimitiveJavaObject(arguments.get()));
　　
                }
　　

　　
                ps.execute();
　　
                ps.close();
　　
                result.set(0);
　　
            } catch (SQLException e) {
　　
                LOG.error("Underlying SQL exception", e);
　　
                result.set(1);
　　
            } finally {
　　
                try {
　　
                    connection.close();
　　
                } catch (Exception ex) {
　　
                    LOG.error("Underlying SQL exception during close", ex);
　　
                }
　　
            }
　　
        }
　　

　　
        return result;
　　
    }
　　

　　
    @Override
　　
    //其用于Hadoop task内部，在使用到这个函数时来展示调试信息
　　
    public String getDisplayString(String[] children) {
　　
        StringBuilder sb = new StringBuilder();
　　
        sb.append("dboutput(");
　　
        if (children.length > 0) {
　　
            sb.append(children[0]);
　　
            for (int i = 1; i < children.length; i++) {
　　
                sb.append(",");
　　
                sb.append(children);
　　
            }
　　
        }
　　

　　
        sb.append(")");
　　
        return sb.toString();
　　
    }
　　
}