Hive 查询 executeQuery() 在 java JDBC 代码中挂起

Question

我创建了一个 UDTF，并在其中运行 java hive JDBC 代码以执行 hive 查询并获取结果。 我能够成功连接到 hive2 服务器，但代码在statement.executeQuery() 处无限期挂起，没有任何异常。可能是什么原因？相同的代码在独立的 eclipse 类中运行，但在作为 udtf 部署在 hadoop 集群中时手。

---

public class DynamicWhereUDTF extends GenericUDTF {
    private PrimitiveObjectInspector stringOI = null;
    ArrayList<Object[]> results = new ArrayList<Object[]>();

    @Override
    public StructObjectInspector initialize(ObjectInspector[] args)
            throws UDFArgumentException {

        stringOI = (PrimitiveObjectInspector) args[0];
        if (stringOI != null) {
            String name = stringOI.toString();
            System.out.println("param <-------> " + name);
        }

        List<String> fieldNames = new ArrayList<String>();
        try {
            fieldNames = getColumnNames("d_drug");
        } catch (SQLException e) {
            e.printStackTrace();
        }
        System.out.println("fieldNames size ---> " + fieldNames.size());
        List<ObjectInspector> fieldOIs = new ArrayList<ObjectInspector>();

        for (int i = 0; i < fieldNames.size(); i++) {
            fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        }

        System.out.println("----------ObjectInspectorFactory created------------ ");
        Connection conn = null;
        ResultSet rs = null;
        PreparedStatement statement = null;
        try {
            System.out.println("Processing records 1");
            Class.forName("org.apache.hive.jdbc.HiveDriver");
            System.out.println("Processing records 2");
            Configuration conf = new Configuration();
            conf.set("hadoop.security.authentication", "Kerberos");
            conf.set("fs.hdfs.impl", DistributedFileSystem.class.getName());
            conf.set("fs.file.impl", org.apache.hadoop.fs.LocalFileSystem.class.getName());
            UserGroupInformation.setConfiguration(conf);
            UserGroupInformation.loginUserFromKeytab("abc@CS.MSD", "/tmp/abc.keytab");
            System.out.println("Processing records 3");
            String hiveJdbcUrl = "jdbc:hive2://<host>:10000/demo_db;principal=hive/<host>@CS.MSD";
            conn = DriverManager.getConnection(hiveJdbcUrl, "abc", "");
            System.out.println("conn1 <-------> " + conn);
            statement = conn.prepareStatement("select * from xyz limit 5");
            System.out.println(" statement ----------> " + statement);
            rs = statement.executeQuery();
            System.out.println(" resultset ----------> " + rs);
            ResultSetMetaData rsMetaData = rs.getMetaData();
            int columnCount = rsMetaData.getColumnCount();
            System.out.println("columnCount ---> " + columnCount);
            // ArrayList<Object[]> results = new ArrayList<Object[]>();
            StringBuilder values = new StringBuilder();

            while (rs.next()) {
                values = new StringBuilder();
                for (int i = 0; i < columnCount; i++) {
                    values = values.append(rs.getString(i + 1)).append(",");
                }
                String output = values.toString().substring(0,
                        values.lastIndexOf(","));
                System.out.println("output  -----> " + output);
                results.add(new Object[] {"122556", "52905"});
            }
            System.out.println("------- results forwarded -------");

        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            if (conn != null)
                try {
                    conn.close();
                } catch (SQLException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
        }
        return ObjectInspectorFactory.getStandardStructObjectInspector(
                fieldNames, fieldOIs);
    }

    @Override
    public void close() throws HiveException {
        // TODO Auto-generated method stub

    }

    @Override
    public void process(Object[] record) throws HiveException {

        try {
            Iterator<Object[]> it = results.iterator();
            while (it.hasNext()) {
                Object[] r = it.next();
                forward(r);
            }
            System.out.println("------- results forwarded -------");
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

    public List<String> getColumnNames(String tableName) throws SQLException {
        List<String> fieldNames = new ArrayList<String>();
        fieldNames.add("drug_id");
        fieldNames.add("drug_cd");
        return fieldNames;
    }
}

Answer 1

问题可能是在initialize 方法中创建连接。尝试用configure方法创建连接，可以以Hbase connector为例。