大数据工具封装.zip

package org.bd.impala; import java.sql.Connection; import java.sql.ResultSet; import java.sql.ResultSetMetaData; import java.sql.SQLException; import java.sql.Statement; import java.util.ArrayList; import java.util.Iterator; import java.util.List; import org.apache.commons.lang.StringUtils; import org.bd.impala.utils.ImpaIaTableColumn; import org.bd.impala.utils.ImpalaConnection; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * <b>版权信息:</b> 2017,big data module<br/> * <b>功能描述:</b> Impala的基本操作<br/> * <b>版本历史:</b><br/> * @author jasse | 2017年3月7日|创建 */ @Deprecated public class ImpalaCommonDao{ private static Logger logger = LoggerFactory.getLogger(ImpalaCommonDao.class); private static ImpalaCommonDao commonDao = null; private Connection conn = null; private Statement stmt = null; private ResultSet rs = null; public ImpalaCommonDao(){ } public static ImpalaCommonDao getInstance(){ try { if(commonDao == null || commonDao.conn == null || commonDao.conn.isClosed()){ commonDao = new ImpalaCommonDao(); } } catch (SQLException e) { logger.error("实例ImpalaCommonDao异常",e.getMessage()); } return commonDao; } /**获取Impala连接*/ public Connection getConn(){ try { if(conn == null || conn.isClosed()){ conn = ImpalaConnection.getInstance().getConn(); } } catch (SQLException e) { logger.error("获取impala连接异常:{0}",e.getMessage()); } return conn; } /** 关闭impala连接*/ public void close(){ try {rs.close();} catch (Exception e2) { logger.error("ResultSet.close()异常:{0}",e2.getMessage()); } try {stmt.close();} catch (Exception e2) { logger.error("Statement.close()异常:{0}",e2.getMessage()); } try {conn.close();} catch (Exception e2) { logger.error("Connection.close()异常:{0}",e2.getMessage()); } } /** * impala表文件格式的枚举 */ public enum TabFileFormat {PARQUET,TEXTFILE,AVRO,SEQUENCEFILE,RCFILE}; /** * 根据sql语句查询结果集合 * @param sql * @return ResultSet * @throws SQLException */ public ResultSet queryData(String sql) throws SQLException { stmt = getConn().createStatement(); rs = stmt.executeQuery(sql); return rs; } /** * 执行DDL语句 * @param sql * @return * @throws SQLException */ public int executeSQL(CharSequence sql) throws SQLException { stmt = getConn().createStatement(); return stmt.executeUpdate(sql.toString()); } /** * 根据表名统计全表数据量 * @param tableName * @return * @throws SQLException */ public int queryCount(String tableName) throws SQLException { try { String sql = "select count(*) from "+tableName; rs = queryData(sql); ResultSetMetaData metaData= rs.getMetaData(); while(rs.next()){ String colname = metaData.getColumnName(1); int count = rs.getInt(colname); return count; } } catch (Exception e) { logger.error("根据表名{0}统计全表数据量,异常:{1}",tableName,e.getMessage()); throw new SQLException("根据表名"+tableName+"统计全表数据量异常!"); }finally{ close(); } return 0; } /** * 根据sql统计数据量 * @param sql 例如：select count(1) from tableA * @return * @throws SQLException */ public int queryCountBySql(String sql) throws SQLException { try { rs = queryData(sql); ResultSetMetaData metaData= rs.getMetaData(); while(rs.next()){ String colname = metaData.getColumnName(1); int count = rs.getInt(colname); return count; } } catch (Exception e) { logger.error("根据sql={0};统计全表数据量,异常:{1}",sql,e.getMessage()); throw new SQLException("根据sql="+sql+";统计数据量异常!"); }finally{ close(); } return 0; } /** * 检查表是否存在 * @param tableName tableA 或者 database.tableA * @return * @throws SQLException */ public boolean checkTable(String tableName) throws SQLException { try { if(StringUtils.isEmpty(tableName)) return false; tableName = tableName.toLowerCase();//Impala默认表名及字段名小写 String sql = "show tables"; if(tableName.contains(".")){ String[] sp = tableName.split("\\."); tableName = sp[1]; sql += " in " + sp[0]; } sql += " like '" + tableName + "'"; rs = queryData(sql); if(rs.next() && tableName.equals(rs.getString(1))) return true; } catch (Exception e) { logger.error("根据表名={0},检查是否存在,异常:{1}",tableName,e.getMessage()); throw new SQLException("根据表名="+tableName+",检查是否存在异常!"); }finally{ close(); } return false; } /** * 判断Impala表字段是否存在 * @param tableName 表名 * @param column 列名 * @return * @throws SQLException */ public boolean checkColumn(String tableName, String column) throws SQLException{ if(StringUtils.isEmpty(tableName) || StringUtils.isEmpty(column)) return false; if(getFields(tableName).contains(column.toLowerCase().trim()))//Impala默认表名及字段名小写 return true; return false; } /** * 获取impala库名列表 * @return * @throws SQLException */ public List<String> getDataBases() throws SQLException { List<String> list = new ArrayList<String>(); String sql = "show databases"; try { rs = queryData(sql); while(rs.next()){ list.add(rs.getString(1)); } } catch (SQLException e) { logger.error("获取impala库名列表异常!"); throw new SQLException("获取impala库名列表异常!"); } return list; } /** * 获取Impala数据库下全部表名 * @param databaseName 库名称 如果查默认库的，则传null 或者"" 值 * @return * @throws SQLException */ public List<String> getTables(String databaseName) throws SQLException { List<String> list = new ArrayList<String>(); try { String sql = "show tables"; if(StringUtils.isNotEmpty(databaseName) && StringUtils.isNotBlank(databaseName)){ sql += " in " + databaseName; } rs = queryData(sql); while(rs.next()){ list.add(rs.getString(1)); } } catch (Exception e) { logger.error("根据库名={0}获取Impala数据库下全部表名异常:{1}",databaseName,e.getMessage()); throw new SQLException("根据库名="+databaseName+",获取Impala数据库下全部表名异常"); }finally{ close(); } return list; } /** * 查询该库下有多少张表 * @param databaseName 库名 * @return * @throws SQLException */ public int getTableCountByDataBase(String databaseName) throws SQLException { int count = 0; try { List<String> list = getTables(databaseName); if (list != null && list.size() > 0) { count = list.size(); } } catch (SQLException e) { logger.error("根据库名={0}查询该库下有多少张表异常:{1}",databaseName,e.getMessage()); throw new SQLException("根据库名="+databaseName+",查询该库下有多少张表异常"); } return count; } /** * 根据表名获取全部字段名 * @param tableName 表名 （例如：tableA 或者 database.tableA） * @return * @throws SQLException */ public List<String> getFields(String tableName) throws SQLException { List<String> list = new ArrayList<String>(); try { if(StringUtils.isEmpty(tableName)) return list; rs = queryData("de