hadoop map reduce hbase 一人一档

package com.aeye.vias.mr.client; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.*; import org.apache.hadoop.hbase.client.*; import org.apache.hadoop.hbase.util.Bytes; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; import java.util.*; import java.util.Map.Entry; public class HbaseClient { private static Logger logger=LoggerFactory.getLogger(HbaseClient.class); public static final String ROWKEY="rowkey"; public Configuration configuration; private volatile static HbaseClient hbaseClient; public static HbaseClient getInstance() { if (hbaseClient == null) { synchronized (HbaseClient.class) { if (hbaseClient == null) { hbaseClient = new HbaseClient(); } } } return hbaseClient; } public HbaseClient(){ configuration= HBaseConfiguration.create(); } /** * 获取连接 * @return * @throws IOException */ public Connection getConnection() throws IOException{ return ConnectionFactory.createConnection(configuration); } /** * 如果表不存在则创建表 * @param tableName 表名 * @param familys 列族列表 * @return 如果表不存在创建了表返回true，表已经存在没有创建表则返回false * @throws Exception */ public boolean createTableIfNotExists(String tableName, String[] familys) throws Exception{ boolean created=false; Connection connection = null; Admin admin = null; try{ connection = getConnection(); admin = connection.getAdmin(); HTableDescriptor table = new HTableDescriptor(TableName.valueOf(tableName)); if(!admin.tableExists(table.getTableName())){ for(String family:familys){ table.setConfiguration(HTableDescriptor.SPLIT_POLICY, "org.apache.hadoop.hbase.regionserver.KeyPrefixRegionSplitPolicy"); table.setConfiguration("KeyPrefixRegionSplitPolicy.prefix_length", "6"); table.addFamily(new HColumnDescriptor(family)); } admin.createTable(table); created=true; logger.info("create table("+tableName+").familys("+Arrays.asList(familys)+") success"); }else{ logger.info("table("+tableName+") already exists,don't need to create"); } }catch (Exception e) { logger.error("Exception:",e); throw e; }finally{ if(admin!=null){ try{ admin.close(); }catch (Exception e) { logger.error("Exception:",e); } } if(connection!=null){ try{ connection.close(); }catch (Exception e) { logger.error("Exception:",e); } } } return created; } /** * 清空表内所有数据 * @param tableName * @throws Exception */ public void truncateTable(String tableName) throws Exception{ Connection connection = null; Admin admin = null; try{ connection = getConnection(); admin = connection.getAdmin(); HTableDescriptor table = new HTableDescriptor(TableName.valueOf(tableName)); if(admin.tableExists(table.getTableName())){ if(admin.isTableEnabled(table.getTableName())){ admin.disableTable(table.getTableName()); } admin.truncateTable(table.getTableName(),false); logger.info("truncate table("+tableName+") success"); }else{ logger.info("table("+tableName+") not exists,don't need to truncate"); } }catch (Exception e) { logger.error("Exception:",e); throw e; }finally{ if(admin!=null){ try{ admin.close(); }catch (Exception e) { logger.error("Exception:",e); } } if(connection!=null){ try{ connection.close(); }catch (Exception e) { logger.error("Exception:",e); } } } } /** * 插入一条数据 * @param tableName 表名 * @param record 待插入的数据 * @throws Exception */ public void insert(String tableName,Map<String,String> record) throws Exception{ List<Map<String,String>> list=new ArrayList<Map<String,String>>(); list.add(record); insertAll(tableName, list); } /** * 插入多条数据 * @param tableName 表名 * @param list 待插入的数据列表 * @throws Exception */ public void insertAll(String tableName,List<Map<String,String>> list) throws Exception{ Connection connection = null; Table table = null; try{ connection = getConnection(); table = connection.getTable(TableName.valueOf(tableName)); Put put=null; List<Put> resultList=new ArrayList<Put>(); for(Map<String,String> map:list){ String rowkey=map.remove(ROWKEY).toString(); put=new Put(Bytes.toBytes(rowkey)); for(Entry<String, String> e:map.entrySet()){ String[] cells=e.getKey().split(":"); put.addColumn(Bytes.toBytes(cells[0]), Bytes.toBytes(cells[1]), Bytes.toBytes(e.getValue())); } resultList.add(put); } table.put(resultList); }catch (Exception e) { logger.error("Exception:",e); throw e; }finally{ if(table!=null){ try{ table.close(); }catch (Exception e) { logger.error("Exception:",e); } } if(connection!=null){ try{ connection.close(); }catch (Exception e) { logger.error("Exception:",e); } } } } /** * 查询数据 * @param tableName 表名 * @param columns 查询的字段名列表，List<String[]{family,qualifier}>，null表示查询所有字段 * @param startRow 开始的rowkey * @param stopRow 结束的rowkey，不包括此rowkey * @param limit 只查询前多少行，0表示查询所有数据 * @return * @throws Exception */ public List<Map<String,String>> query(String tableName,List<String[]> columns,String startRow,String stopRow,int limit) throws Exception{ List<Map<String,String>> list = null; Connection connection = null; Table table = null; ResultScanner resultScanner = null; try{ connection = getConnection(); table = connection.getTable(TableName.valueOf(tableName)); Scan scan=new Scan(Bytes.toBytes(startRow), Bytes.toBytes(stopRow)); scan.setCaching(limit>0 && limit<100?limit:100); if(limit>0){ scan.setMaxResultSize(limit); } if(columns!=null && columns.size()>0){ for(String[] column:columns){ scan.addColumn(Bytes.toBytes(column[0]), Bytes.toBytes(column[1])); } } resultScanner = table.getScanner(scan); list=new ArrayList<Map<String,String>>(); for(Result result:resultScanner){ Map<String,String> row=new LinkedHashMap<String,String>(); row.put(ROWKEY, Bytes.toString(result.getRow())); for(Cell cell:result.listCells()){ row.put(Bytes.toString(CellUtil.cloneFamily(cell))+":"+Bytes.toString(CellUtil.cloneQualifier(cell)), Bytes.toString(CellUtil.cloneValue(cell))); } list.add(row); } }catch (Exception e) { logger.error("Exception:",e); throw e; }finally{ if(resultScanner!=null){ try{ resultScanner.close(); }catch (Exception e) { logger.error("Exception:",e); } } if(table!=null){ try{ table.close(); }catch (Exception e) { logger.error("Exception:",e); } } if(connection!=null){ try{ connection.close(); }catch (Exception e) { logger.error("Exception:",e); } } } return list; } /** * 查询数据 * @param tableName 表名 * @param columns 查询的字段名列表，List<String[]{family,qualifier}>，null表示查询所有字段 * @param limit 只查询前多少行，0表示查询所有数