HBase表操作

相对于0.9.X版本，在HBase1.X版本对内部API改动比较大，例如连接部分类库变更，如下：

　　连接获取：org.apache.hadoop.hbase.HBaseConfiguration.createConnection(conf)已经过时，改为使用org.apache.hadoop.hbase.client.ConnectionFactory.createConnection(conf)；

　　表操作：org.apache.hadoop.hbase.client.HTable已过时，改为使用org.apache.hadoop.hbase.client.HTable接口；

　　

第一部分：连接获取
package com.mengyao.bigdata.hbase;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;

/**
 * 
 * @author mengyao
 *
 */
public class HBaseHelper {

    private static final String QUORUM = "hbase.zookeeper.quorum";
    private static final String ZK_PORT = "hbase.zookeeper.property.clientPort";
    private String quorum;
    private String zkPort;
    private Configuration conf;

    public HBaseHelper(Configuration conf) {
        this.conf = conf;
    }
    
    public HBaseHelper(String quorum, String zkPort) {
        this.quorum = quorum;
        this.zkPort = zkPort;
    }
    
    public HBaseHelper(Configuration conf, String quorum, String zkPort) {
        this.conf = conf;
        this.quorum = quorum;
        this.zkPort = zkPort;
    }
    
    /**
     * 
     * @param conf
     * @return
     * @throws IOException
     */
    public Connection getConnection() throws IOException {
        if (null == conf) {
            conf = HBaseConfiguration.create();
            conf.set(QUORUM, quorum);
            conf.set(ZK_PORT, zkPort);
        }
        if (null == conf.get(QUORUM) || conf.get(QUORUM).isEmpty()) {
            conf.set(QUORUM, quorum);            
        }
        if (null == conf.get(ZK_PORT) || conf.get(ZK_PORT).isEmpty()) {
            conf.set(ZK_PORT, zkPort);
        }
        return ConnectionFactory.createConnection(conf);
    }
    
    /**
     * 
     * @param connection
     * @throws IOException
     */
    public void closeAll(Connection connection) throws IOException {
        if (!connection.isClosed()) {
            connection.close();
        }
    }
    
    public String getQuorum() {
        return quorum;
    }

    public void setQuorum(String quorum) {
        this.quorum = quorum;
    }

    public String getZkPort() {
        return zkPort;
    }

    public void setZkPort(String zkPort) {
        this.zkPort = zkPort;
    }

}



第二部分：表数据增删改查
package com.mengyao.bigdata.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;

/**
 * HBase version:1.0.1.1
 * @author mengyao
 *
 */
public class HBaseV1011Template {

    Logger logger = Logger.getLogger(getClass());

    private static Connection connection;
    static {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "bdata200,bdata202,bdata203");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        HBaseHelper hBaseHelper = new HBaseHelper(conf);
        try {
            connection = hBaseHelper.getConnection();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws IOException {
        //单行添加
        Map<String, String> keyValue = new HashMap<>();
        keyValue.put("name", "xiaogang");
        keyValue.put("sex", "男");
        keyValue.put("addr", "北京");
        add("test", "0005", "info", keyValue);
        
        //多行添加
        Map<String, Map<String, String>> keyValues = new HashMap<String, Map<String, String>>();
        Map<String, String> keyValue1 = new HashMap<>();
        keyValue1.put("name", "ligang");
        keyValue1.put("sex", "男");
        keyValue1.put("addr", "北京");
        keyValues.put("0008_info", keyValue1);
        Map<String, String> keyValue2 = new HashMap<>();
        keyValue2.put("name", "zhaojun");
        keyValue2.put("sex", "男");
        keyValue2.put("addr", "北京");
        keyValues.put("0009_info", keyValue2);
        adds("test", "_", keyValues);

        //查询所有
        System.out.println("1:"+queryForScan("test", "info"));
        
        //根据rowKey查询
        System.out.println("2:"+queryForRowKey("test", "0005", "info"));
        
        //根据时间戳范围查询（默认包钱不包后）
        System.out.println("3:"+queryForTimeRange("test", "info", 1492510703521L, 1492664183955L));
        
        //根据rowKey查询（默认包前不包后）
        System.out.println("4:"+queryForRowKeyRange("test", "info", "0001", "0003"));
        
        //根据指定列名和列值查询1
        System.out.println("5:"+queryForQuilfier("test", "info", "name", "xiaoming"));
        
        //根据指定列名和列值查询1
        System.out.println("6:"+queryForQuilfier("test", "info", "sex", "男"));
        
        //根据指定列名和列值查询1
        System.out.println("7:"+queryForQuilfier("test", "info", "sex", "女"));
        
        //根据rowKey删除
        deleteByRowKey("test", "0005", "info");
    }

    /**
     * 单行添加
     * @param tableName
     * @param rowKey
     * @param family
     * @param keyValue
     * @throws IOException
     */
    private static void add(String tableName, String rowKey, String family, Map<String, String> keyValue) throws IOException {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Put put = new Put(Bytes.toBytes(rowKey));
        for (Entry<String, String> entry : keyValue.entrySet()) {
            put.addColumn(Bytes.toBytes(family), Bytes.toBytes(entry.getKey()), Bytes.toBytes(entry.getValue()));
        }
        table.put(put);
        table.close();
        keyValue.clear();
    }
    
    
    private static void adds(String tableName, String rowFamilySeparator, Map<String, Map<String, String>> keyValues) throws IOException {
        Table table = connection.getTable(TableName.valueOf(tableName));
        List<Put> puts = new ArrayList<Put>();
        for (Entry<String, Map<String, String>> entry : keyValues.entrySet()) {
            String key = entry.getKey();
            if (null == rowFamilySeparator || rowFamilySeparator.isEmpty()) {
                rowFamilySeparator = "_";
            }
            String rowKey = key.split(rowFamilySeparator)[0];
            String family = key.split(rowFamilySeparator)[1];
            Map<String, String> keyValue = entry.getValue();
            Put put = new Put(Bytes.toBytes(rowKey), System.currentTimeMillis());
            for (Entry<String, String> entry2 : keyValue.entrySet()) {
                put.addColumn(Bytes.toBytes(family), Bytes.toBytes(entry2.getKey()), Bytes.toBytes(entry2.getValue()));                
            }
            puts.add(put);
        }
        table.put(puts);
        table.close();
        keyValues.clear();
    }
    
    /**
     * 单行删除
     * @param tableName
     * @param rowKey
     * @param family
     * @throws IOException
     */
    private static void deleteByRowKey(String tableName, String rowKey, String family) throws IOException {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Delete delete = new Delete(Bytes.toBytes(rowKey)); 
        table.delete(delete);
        table.close();
    }

    /**
     * 查询所有
     * @param tableName
     * @param family
     * @return
     * @throws IOException
     */
    public static List<Map<String, String>> queryForScan(String tableName, String family) throws IOException {
        List<Map<String, String>> rows = new ArrayList<Map<String, String>>();
        Table table = connection.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        scan.addFamily(Bytes.toBytes(family));
        ResultScanner rs = table.getScanner(scan);
        Map<String, String> row = null;
        try {
            for (Result r = rs.next(); r != null; r = rs.next()) {
                Cell[] cells = r.rawCells();
                for (Cell cell : cells) {
                    row = new HashMap<String, String>();
                    row.put("timestamp", cell.getTimestamp() + "");
                    row.put("rowKey", new String(CellUtil.cloneRow(cell)));
                    row.put("family", new String(CellUtil.cloneFamily(cell)));
                    row.put(new String(CellUtil.cloneQualifier(cell)), new String(CellUtil.cloneValue(cell)));
                    rows.add(row);
                }
            }
        } finally {
            rs.close();
        }
        return rows;
    }
    
    /**
     * 根据时间范围
     * @param tableName
     * @param family
     * @param minStamp
     * @param maxStamp
     * @return
     * @throws IOException
     */
    public static List<Map<String, String>> queryForTimeRange(String tableName, String family, long minStamp, long maxStamp) throws IOException {
        List<Map<String, String>> rows = new ArrayList<Map<String, String>>();
        Table table = connection.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        scan.addFamily(Bytes.toBytes(family));
        scan.setTimeRange(minStamp, maxStamp);
        ResultScanner rs = table.getScanner(scan);
        Map<String, String> row = null;
        try {
            for (Result r = rs.next(); r != null; r = rs.next()) {
                Cell[] cells = r.rawCells();
                for (Cell cell : cells) {
                    row = new HashMap<String, String>();
                    row.put("timestamp", cell.getTimestamp() + "");
                    row.put("rowKey", new String(CellUtil.cloneRow(cell)));
                    row.put("family", new String(CellUtil.cloneFamily(cell)));
                    row.put(new String(CellUtil.cloneQualifier(cell)), new String(CellUtil.cloneValue(cell)));
                    rows.add(row);
                }
            }
        } finally {
            rs.close();
        }
        return rows;
    }

    /**
     * 根据RowKey查询
     * @param tableName
     * @param rowKey
     * @param family
     * @return
     * @throws IOException
     */
    public static Map<String, String> queryForRowKey(String tableName, String rowKey, String family) throws IOException {
        Table table = connection.getTable(TableName.valueOf(tableName));
        Get get = new Get(Bytes.toBytes(rowKey));
        get.addFamily(Bytes.toBytes(family));
        Scan scan = new Scan(get);
        ResultScanner rs = table.getScanner(scan);
        Map<String, String> row = null;
        try {
            for (Result r = rs.next(); r != null; r = rs.next()) {
                Cell[] cells = r.rawCells();
                row = new HashMap<String, String>();
                for (Cell cell : cells) {
                    row.put("timestamp", cell.getTimestamp() + "");
                    row.put("rowKey", new String(CellUtil.cloneRow(cell)));
                    row.put("family", new String(CellUtil.cloneFamily(cell)));
                    row.put(new String(CellUtil.cloneQualifier(cell)), new String(CellUtil.cloneValue(cell)));
                }
            }
        } finally {
            rs.close();
        }
        return row;
    }
    
    /**
     * 根据RowKey范围查询
     * @param tableName
     * @param family
     * @param startRow
     * @param stopRow
     * @return
     * @throws IOException
     */
    public static List<Map<String, String>> queryForRowKeyRange(String tableName, String family, String startRow, String stopRow) throws IOException {
        List<Map<String, String>> rows = new ArrayList<Map<String, String>>();
        Table table = connection.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        scan.addFamily(Bytes.toBytes(family));
        scan.setStartRow(Bytes.toBytes(startRow));
        scan.setStopRow(Bytes.toBytes(stopRow));
        ResultScanner rs = table.getScanner(scan);
        Map<String, String> row = null;
        try {
            for (Result r = rs.next(); r != null; r = rs.next()) {
                Cell[] cells = r.rawCells();
                for (Cell cell : cells) {
                    row = new HashMap<String, String>();
                    row.put("timestamp", cell.getTimestamp() + "");
                    row.put("rowKey", new String(CellUtil.cloneRow(cell)));
                    row.put("family", new String(CellUtil.cloneFamily(cell)));
                    row.put(new String(CellUtil.cloneQualifier(cell)), new String(CellUtil.cloneValue(cell)));
                    rows.add(row);
                }
            }
        } finally {
            rs.close();
        }
        return rows;
    }
    
    /**
     * 根据指定列名匹配列值
     * @param tableName
     * @param family
     * @param qualifier
     * @param value
     * @return
     * @throws IOException
     */
    public static Collection<Map<String, String>> queryForQuilfier(String tableName, String family, String qualifier, String value) throws IOException {
        Map<String, Map<String, String>> rows = new HashMap<String, Map<String, String>>();
        Table table = connection.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        SingleColumnValueFilter filter = new SingleColumnValueFilter(Bytes.toBytes(family), Bytes.toBytes(qualifier), CompareOp.EQUAL, Bytes.toBytes(value));
        filter.setFilterIfMissing(true); 
        scan.setFilter(filter);
        ResultScanner rs = table.getScanner(scan);
        Map<String, String> row = null;
        try {
            for (Result r = rs.next(); r != null; r = rs.next()) {
                Cell[] cells = r.rawCells();
                for (Cell cell : cells) {
                    String rowKey = new String(CellUtil.cloneRow(cell));
                    if (null == row || !rows.containsKey(rowKey)) {
                        row = new HashMap<String, String>();
                    }
                    row.put("timestamp", cell.getTimestamp() + "");
                    row.put("rowKey", rowKey);
                    row.put("family", new String(CellUtil.cloneFamily(cell)));
                    row.put(new String(CellUtil.cloneQualifier(cell)), new String(CellUtil.cloneValue(cell)));
                    rows.put(rowKey,row);
                }
            }
        } finally {
            rs.close();
        }
        return rows.values();
    }

}