Apache Flink JDBC InputFormat 抛出 java.net.SocketException: Socket closed答案

【问题标题】：Apache Flink JDBC InputFormat throwing java.net.SocketException: Socket closedApache Flink JDBC InputFormat 抛出 java.net.SocketException: Socket closed
【发布时间】：2016-08-26 04:00:29
【问题描述】：

我正在使用 Flink DataSet API 查询 oracle 数据库。为此，我自定义了 Flink JDBCInputFormat 以返回 java.sql.Resultset。因为我需要使用 Flink 运算符对结果集执行进一步的操作。

public static void main(String[] args) throws Exception { 

    ExecutionEnvironment environment = ExecutionEnvironment.getExecutionEnvironment();
    environment.setParallelism(1);
    @SuppressWarnings("unchecked")
    DataSource<ResultSet> source
            = environment.createInput(JDBCInputFormat.buildJDBCInputFormat()
                    .setUsername("username")
                    .setPassword("password")
                    .setDrivername("driver_name")
                    .setDBUrl("jdbcUrl")
                    .setQuery("query")
                    .finish(),      
                    new GenericTypeInfo<ResultSet>(ResultSet.class)
            );
    source.print();

    environment.execute();

}

以下是自定义的JDBCInputFormat：

public class JDBCInputFormat extends RichInputFormat<ResultSet, InputSplit> implements ResultTypeQueryable {

@Override
public void open(InputSplit inputSplit) throws IOException {
                Class.forName(drivername);
                    dbConn = DriverManager.getConnection(dbURL, username, password);
                statement = dbConn.prepareStatement(queryTemplate, resultSetType, resultSetConcurrency);
                resultSet = statement.executeQuery();
}

@Override
public void close() throws IOException {
            if(statement != null) {
                    statement.close();
                }
                if(resultSet != null) 
                    resultSet.close();
                if(dbConn != null) {
                    dbConn.close();
                }
}

@Override
public boolean reachedEnd() throws IOException {
        isLastRecord = resultSet.isLast();
    return isLastRecord;
}

@Override
public ResultSet nextRecord(ResultSet row) throws IOException{  
        if(!isLastRecord){              
            resultSet.next();
        }
        return resultSet;
}

}

这适用于在获取的行中有限制的以下查询： SELECT a,b,c from xyz where rownum 但是当我尝试获取所有行拥有大约 100 万条数据时，我在获取随机行数后得到以下异常：

java.sql.SQLRecoverableException: Io exception: Socket closed
at oracle.jdbc.driver.SQLStateMapping.newSQLException(SQLStateMapping.java:101)
at oracle.jdbc.driver.DatabaseError.newSQLException(DatabaseError.java:133)
at oracle.jdbc.driver.DatabaseError.throwSqlException(DatabaseError.java:199)
at oracle.jdbc.driver.DatabaseError.throwSqlException(DatabaseError.java:263)
at oracle.jdbc.driver.DatabaseError.throwSqlException(DatabaseError.java:521)
at oracle.jdbc.driver.T4CPreparedStatement.fetch(T4CPreparedStatement.java:1024)
at oracle.jdbc.driver.OracleResultSetImpl.close_or_fetch_from_next(OracleResultSetImpl.java:314)
at oracle.jdbc.driver.OracleResultSetImpl.next(OracleResultSetImpl.java:228)
at oracle.jdbc.driver.ScrollableResultSet.cacheRowAt(ScrollableResultSet.java:1839)
at oracle.jdbc.driver.ScrollableResultSet.isValidRow(ScrollableResultSet.java:1823)
at oracle.jdbc.driver.ScrollableResultSet.isLast(ScrollableResultSet.java:349)
at JDBCInputFormat.reachedEnd(JDBCInputFormat.java:98)
at org.apache.flink.runtime.operators.DataSourceTask.invoke(DataSourceTask.java:173)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:559)
at java.lang.Thread.run(Thread.java:745)

原因：java.net.SocketException：套接字关闭在 java.net.SocketOutputStream.socketWrite0(Native Method)

那么对于我的情况，我该如何解决这个问题？

【问题讨论】：

标签： apache-flink flink-streaming

【解决方案1】：

我认为不可能像普通唱片那样发送ResultSet。这是一个有状态的对象，它在内部维护与数据库服务器的连接。使用ResultSet 作为在 Flink 操作员之间传输的记录意味着它可以被序列化、通过网络传送到另一台机器、反序列化并交给不同 JVM 进程中的不同线程。那是行不通的。

根据连接的不同，ResultSet 也可以留在同一台机器上的同一线程中，这可能对您有用。如果您想从运算符中查询数据库，您可以将该函数实现为RichMapPartitionFunction。否则，我会读取数据源中的ResultSet 并转发结果行。

【讨论】：