impala jdbc驅動執行impala sql的一個坑（不支援多行sql）

架構使用spark streaming 消費kafka的資料，並通過impala來插入到kudu中，但是通過對比發現落地到kudu表中的資料比kafka訊息數要少，通過後臺日志發現，偶發性的出現java.sql.SQLException: [Simba][ImpalaJDBCDriver](500051) ERROR processing query/statement. Error Code: 0, SQL state: TStatus(statusCode:ERROR_STATUS, sqlState:HY000, errorMessage:AnalysisException: Syntax error in line 1

原因是呼叫過程中使用了資料庫連線池,會合並多行sql執行，實現如下：

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidDataSourceFactory;
import com.alibaba.druid.pool.DruidPooledConnection;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.sql.SQLException;
import java.util.Properties;


public class ImapalConnPool {

private static Log logger = LogFactory.getLog(ImapalConnPool.class);

private static ImapalConnPool imapalConnPool = null;
private static DruidDataSource druidDataSource = null;

static {
Properties properties = new Properties();
properties.setProperty("driverClassName","com.cloudera.impala.jdbc41.Driver");
properties.setProperty("url","jdbc:impala://127.0.0.1:21050");
properties.setProperty("username","");
properties.setProperty("password","");
properties.setProperty("initialSize","50");
properties.setProperty("maxActive","100");
properties.setProperty("maxWait","60000");
properties.setProperty("timeBetweenEvictionRunsMillis","60000");
properties.setProperty("minEvictableIdleTimeMillis","300000");
properties.setProperty("validationQuery","SELECT 1");
properties.setProperty("testWhileIdle","true");
properties.setProperty("testOnBorrow","false");
properties.setProperty("testOnReturn","false");
properties.setProperty("poolPreparedStatements","false");
//當該值大於0時，啟用pool，poolPreparedStatements為true
properties.setProperty("maxPoolPreparedStatementPerConnectionSize","-1");

try {
druidDataSource = (DruidDataSource) DruidDataSourceFactory.createDataSource(properties); //DruidDataSrouce工廠模式
} catch (Exception e) {
logger.error(e);
}
}

public static ImapalConnPool getInstance(){
if (null == imapalConnPool){
synchronized(ImapalConnPool.class) {
if(null == imapalConnPool) {
imapalConnPool = new ImapalConnPool();
}
}
}
return imapalConnPool;
}

public DruidPooledConnection getConnection() throws SQLException {
return druidDataSource.getConnection();
}

}

調整上述引數poolPreparedStatements和maxPoolPreparedStatementPerConnectionSize任然不能解決同一個connection合併多行sql的問題

Connection connectionn = null;
Statement statement = null;
try {
Class.forName("com.cloudera.impala.jdbc41.Driver");
connectionn = DriverManager.getConnection("jdbc:impala://127.0.0.1:21050");
statement = connectionn.createStatement();
for (String item : execSql) {
statement.execute(item);
}
processResult = true;
}catch (Exception ex){
logger.error(this,ex);
}finally {
if(null != statement){
try {
statement.close();
}catch (Exception ex){
logger.error(this,ex);
}
}
if(null != connectionn) {
try {
connectionn.close();
}catch (Exception ex){
logger.error(this,ex);
}
}
}