java - 在线程中迭代结果集会导致异常。使用 JDBC 连接对 Hive 进行查询的结果集

Question

对 Hive 及其功能进行评估。有一个用例，我需要在单独的线程中迭代结果集。我可以拥有许多结果集并生成一个线程来处理每个结果集。下面是我为此用例编写的代码。

public class ConcurrentRSIteration2 {
private static String[] tableNames = 
    { 
        "random_data1",
        "random_data2",
        "random_data3",
        "random_data4"
    };

public static void main(String args[]) throws Exception {
    String driverName = "org.apache.hive.jdbc.HiveDriver";
    Class.forName(driverName);
    Connection con = DriverManager.getConnection(
            "jdbc:hive2://127.0.0.1:10000/default", "hive", "");

    int length = tableNames.length;

    StringBuilder[] sql = new StringBuilder[length];
    PreparedStatement[] stmt = new PreparedStatement[length];
    Thread[] rsIterators = new Thread[length];

    for (int i = 0; i < length; i++) {
        sql[i] = new StringBuilder().
                    append("select * from ").
                    append(tableNames[i]);
        stmt[i] = con.prepareStatement(sql[i].toString());
        RSIterator2 rsIterator = new RSIterator2(stmt[i].executeQuery());

        rsIterators[i] = new Thread(rsIterator);
    }

    for (int i = 0; i < length; i++) {
        rsIterators[i].start();
    }
}
}

class RSIterator2 implements Runnable {

private ResultSet rs;

public RSIterator2(ResultSet rs) {
    this.rs = rs;
}

@Override
public void run() {
    try {
        System.out.println(this.hashCode() + " : " + rs);
        System.out.println(this.hashCode() + " : RS iteration started.");
        int i = 0;
        while (rs.next()) {
            i++;
        }
        System.out.println(this.hashCode() + " : RS iteration done.");
    } catch (SQLException e) {
        e.printStackTrace();
    } finally {
        try {
            rs.close();
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }
}

}

下面是异常的堆栈跟踪。

org.apache.thrift.transport.TTransportException
    at org.apache.thrift.transport.TIOStreamTransport.read(TIOStreamTransport.java:132)
    at org.apache.thrift.transport.TTransport.readAll(TTransport.java:86)
    at org.apache.thrift.transport.TSaslTransport.readLength(TSaslTransport.java:376)
    at org.apache.thrift.transport.TSaslTransport.readFrame(TSaslTransport.java:453)
    at org.apache.thrift.transport.TSaslTransport.read(TSaslTransport.java:435)
    at org.apache.thrift.transport.TSaslClientTransport.read(TSaslClientTransport.java:37)
    at org.apache.thrift.transport.TTransport.readAll(TTransport.java:86)
    at org.apache.thrift.protocol.TBinaryProtocol.readAll(TBinaryProtocol.java:429)
    at org.apache.thrift.protocol.TBinaryProtocol.readI32(TBinaryProtocol.java:318)
    at org.apache.thrift.protocol.TBinaryProtocol.readMessageBegin(TBinaryProtocol.java:219)
    at org.apache.thrift.TServiceClient.receiveBase(TServiceClient.java:69)
    at org.apache.hive.service.cli.thrift.TCLIService$Client.recv_FetchResults(TCLIService.java:501)
    at org.apache.hive.service.cli.thrift.TCLIService$Client.FetchResults(TCLIService.java:488)
    at org.apache.hive.jdbc.HiveQueryResultSet.next(HiveQueryResultSet.java:360)
    at hivetrial.RSIterator2.run(ConcurrentRSIteration2.java:60)
    at java.lang.Thread.run(Unknown Source)`

我是 Hive 新手，可能忽略了一些事情。试图理解这个异常。

Answer 1

你的整个方法都是建立在一个谬论之上的。您正在使用单个连接来执行多个查询。因此，数据库服务器将按照查询执行的顺序对返回的所有数据进行排序。使用多个线程处理单个流开始没有意义。

您也永远不会关闭语句或连接。