更新时间:2022-07-19 GMT+08:00
Java样例代码
功能简介
使用自定义客户端的JDBC接口提交数据分析任务,并返回结果。
样例代码
- 定义SQL语句。SQL语句必须为单条语句,注意其中不能包含“;”。示例:
ArrayList<String> sqlList = new ArrayList<String>(); sqlList.add("CREATE TABLE CHILD (NAME STRING, AGE INT) ROW FORMAT DELIMITED FIELDS TERMINATED BY ','"); sqlList.add("LOAD DATA INPATH '/home/data' INTO TABLE CHILD"); sqlList.add("SELECT * FROM child"); sqlList.add("DROP TABLE child"); executeSql(url, sqlList);
- 样例工程中的data文件需要放到HDFS上的home目录下
- 保证data文件和创建的表的所属的用户和用户组保持一致
- 拼接JDBC URL。
HA模式下url的host和port必须为“ha-cluster”。
普通集群需要将样例代码中com.huawei.bigdata.spark.examples.ThriftServerQueriesTest类中第67、68行代码
StringBuilder sb = new StringBuilder("jdbc:hive2://ha-cluster/default"
+ securityConfig);
改为StringBuilder sb = new StringBuilder("jdbc:hive2://ha-cluster/default");
String HA_CLUSTER_URL = "ha-cluster"; StringBuilder sb = new StringBuilder("jdbc:hive2://" + HA_CLUSTER_URL + "/default;"); String url = sb.toString();
- 加载Hive JDBC驱动。
Class.forName("org.apache.hive.jdbc.HiveDriver").newInstance();
- 获取JDBC连接,执行HQL,输出查询的列名和结果到控制台,关闭JDBC连接。
在网络拥塞的情况下,您还可以设置客户端与JDBCServer连接的超时时间,可以避免客户端由于无限等待服务端的返回而挂起。使用方式如下:
在执行“DriverManager.getConnection”方法获取JDBC连接前,添加“DriverManager.setLoginTimeout(n)”方法来设置超时时长,其中n表示等待服务返回的超时时长,单位为秒,类型为Int,默认为“0”(表示永不超时)。
static void executeSql(String url, ArrayList<String> sqls) throws ClassNotFoundException, SQLException { try { Class.forName("org.apache.hive.jdbc.HiveDriver").newInstance(); } catch (Exception e) { e.printStackTrace(); } Connection connection = null; PreparedStatement statement = null; try { connection = DriverManager.getConnection(url); for (int i = 0 ; i < sqls.size(); i++) { String sql = sqls.get(i); System.out.println("---- Begin executing sql: " + sql + " ----"); statement = connection.prepareStatement(sql); ResultSet result = statement.executeQuery(); ResultSetMetaData resultMetaData = result.getMetaData(); Integer colNum = resultMetaData.getColumnCount(); for (int j = 1; j < colNum; j++) { System.out.println(resultMetaData.getColumnLabel(j) + "\t"); } System.out.println(); while (result.next()) { for (int j = 1; j < colNum; j++){ System.out.println(result.getString(j) + "\t"); } System.out.println(); } System.out.println("---- Done executing sql: " + sql + " ----"); } } catch (Exception e) { e.printStackTrace(); } finally { if (null != statement) { statement.close(); } if (null != connection) { connection.close(); } } }
父主题: 通过JDBC访问Spark SQL的程序