HBase 常用java api获得客户端,创建表,查询,删除

Posted yanyuechao

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了HBase 常用java api获得客户端,创建表,查询,删除相关的知识,希望对你有一定的参考价值。

1,前期准备

(1) 本文采用的hbase是采用三台服务器搭建的集群,zookeeper也是相同服务器搭建的集群,集群ip分别是192.168.183.101; 192.168.183.102; 192.168.183.103。其中102是主节点(HMaster),101以及103都是HRegionServer

(2) 这次测试安装的hbase的版本是 hbase-0.99.2.-bin.tar

(3)java api引用的maven依赖路径如下

 <dependency>
     <groupId>org.apache.hbase</groupId>
     <artifactId>hbase-client</artifactId>
      <version>1.0.2</version>
</dependency>

(4)配置本地的hosts文件(在本地配置集群的ip与主机名的映射关系)

zookeeper集群的内部有时候通过主机名来进行寻址。如果不在配置hosts文件,在获得下面的Connection连接时,程序一直会在createConenction这个方法上,测试时等待了3分钟程序一直卡着。

C:\Windows\System32\drivers\etc\hosts的配置如下:

192.168.183.101 mini01
192.168.183.102 mini02
192.168.183.103 mini03

 

2,获取Connection对象

引入maven后,先获取hbase的java操作HBase的Connection对象,传入zookeeper的地址以及zookeeper的端口号zookeeper,通过ConnectionFactory可以获取hbase的连接Connection.

    /**
     * 获取Connection对象 
     */
    static Configuration config = null;
    private Connection connection = null;
    private Table table = null;

    Logger LOG = LoggerFactory.getLogger(HbaseGetImpl.class);
    @Before
    public void init() throws Exception {
        config = HBaseConfiguration.create();// 配置
        config.set("hbase.zookeeper.quorum", "mini01,mini02,mini03");// zookeeper地址
        config.set("hbase.zookeeper.property.clientPort", "2181");// zookeeper端口
        connection = ConnectionFactory.createConnection(config);
    }

下面看一下ConnectionFactory.createConnection(config)的方法注解

/**

 * Create a new Connection instance using the passed <code>conf</code> instance. Connection
 * encapsulates all housekeeping for a connection to the cluster. All tables and interfaces
 * created from returned connection share zookeeper connection, meta cache, and connections
 * to region servers and masters.
 * <br>
 * The caller is responsible for calling {@link Connection#close()} on the returned
 * connection instance.
 *  调用方在调用拿到connection之后,有责任在随后的代码中调用Connection.close()方法来关闭连接
 * Typical usage:
   典型用法如下:
 * <pre>
 * Connection connection = ConnectionFactory.createConnection(conf);
 * Table table = connection.getTable(TableName.valueOf("table1"));
 * try {
 *   table.get(...);
 *   ...
 * } finally {
 *   table.close();
 *   connection.close();
 * }
 * </pre>
 *
 * @param conf configuration
 * @param user the user the connection is for
 * @param pool the thread pool to use for batch operations
 * @return Connection object for <code>conf</code>
 */
public static Connection createConnection(Configuration conf, ExecutorService pool, User user)
throws IOException {
  if (user == null) {
    UserProvider provider = UserProvider.instantiate(conf);
    user = provider.getCurrent();
  }

  return createConnection(conf, false, pool, user);
}

 

3,创建hbase的表

利用connection对象,可以指定HBase表的表名以及表的列族,可以调用admin.createTable(htd, splits);的方法来创建Hbase的表。下面说明以下三点注意事项:

(1)hcd.setDataBlockEncoding(DataBlockEncoding.FAST_DIFF)这个方法可以指定hbase数据块的编码类型,可以选择的是DIFF,FAST_DIFF,PREFIX这几种的编码类型。

(2)hcd.setCompressionType的方法,可以指定数据压缩算法,分别是GZ和SNAPPY。值得注意的是在我的本机集群测试时,当我选择SNAPPY作为压缩算法时,在调用testCreateTable时会卡住,并且不能创建表。当我把这行代码注释掉后,创建hbase表顺利。怀疑是我本机的hbase版本与Snappy的配置项不匹配。

(3)在创建表时,admin.createTable(htd, splits)方法可以指定指定splits的参数,预定义表的region分区,这边预定了4个region分区。

(4)在创建hbase表结束后,需要将connection关闭。

@Test
    public void testCreateTable() {

        LOG.info("Entering testCreateTable.");
// Specify the table descriptor.
        HTableDescriptor htd = new HTableDescriptor(TableName.valueOf("student2"));
// Set the column family name to info.
        HColumnDescriptor hcd = new HColumnDescriptor("info");
// Set data encoding methods, HBase provides DIFF,FAST_DIFF,PREFIX
// and PREFIX_TREE
        hcd.setDataBlockEncoding(DataBlockEncoding.FAST_DIFF);
// Set compression methods, HBase provides two default compression
// methods:GZ and SNAPPY
// GZ has the highest compression rate,but low compression and
// decompression effeciency,fit for cold data
// SNAPPY has low compression rate, but high compression and
// decompression effeciency,fit for hot data.
// it is advised to use SNAANPPY
//   hcd.setCompressionType(Compression.Algorithm.SNAPPY);
        htd.addFamily(hcd);
        Admin admin = null; //注[2]
        try {
// Instantiate an Admin object.
            admin = connection.getAdmin();
            if (!admin.tableExists(TableName.valueOf("student2"))) {
                LOG.info("Creating table...");
//                admin.createTable(htd);
                // 创建一个预划分region的表
                byte[][] splits = new byte[4][];
                splits[0] = Bytes.toBytes("A");
                splits[1] = Bytes.toBytes("H");
                splits[2] = Bytes.toBytes("O");
                splits[3] = Bytes.toBytes("U");
                admin.createTable(htd, splits);
                LOG.info(String.valueOf(admin.getClusterStatus()));
                LOG.info(String.valueOf(admin.listNamespaceDescriptors()));
                LOG.info("Table created successfully.");
            } else {
                LOG.warn("table already exists");
            }
        } catch (IOException e) {
            LOG.error("Create table failed " ,e);
        } finally {
            if (admin != null) {
                try {
// Close the Admin object.
                    admin.close();
                } catch (IOException e) {
                    LOG.error("Failed to close admin " ,e);
                }
            }
        }
        LOG.info("Exiting testCreateTable.");
    }

2,测试往hbase表中put数据

(1)先通过connnection以及tableName获得Table对象

(2)再构造Put对象时,Put put = new Put(rowkey),在构造方法里面的是hbase表的唯一键rowkey(相当于mysql中的id)。 此外还要指定两个参数,分别是familyName 列族的名字 ,qulifiers列名。这边需要注意的是tableName以及familyName是在创建表时就已经确定好了的。而hbase中的列名qulifers不同于mysql中的列名,mysql中的列名是在创建mysql表时就已经确定的。qulfiers可以在插入输入时随意指定名称,不是预定义的。

(3)hbase存储的数据是不带数据类型的,全是Bytes。所以在插入hbase表前,需要调用Byte.toBytes()方法,将数据转成Bytes后再进行插入。

 @Test
    public void testPut() {
        // Specify the column family name.
        byte[] familyName = Bytes.toBytes("info");
//        byte[] familyName1 = Bytes.toBytes("");
        // Specify the column name.
        byte[][] qualifiers = {
                Bytes.toBytes("colu1")
//                Bytes.toBytes("l2"),
//                Bytes.toBytes("l3")
        };
        Table table = null;
        try {
            // Instantiate an HTable object.
            table = connection.getTable(TableName.valueOf("student2"));
            List<Put> puts = new ArrayList<Put>();
            for(int i=1;i<10;i=i+1){
                // Instantiate a Put object.
               String rowkey = UUID.randomUUID().toString(); 
Put put = new Put(Bytes.toBytes(rowkey)); //high_risk put.addColumn(familyName, qualifiers[0], Bytes.toBytes("tommy1")); // put.addColumn(familyName, qualifiers[0], Bytes.toBytes(80)); // put.addColumn(familyName, qualifiers[0], Bytes.toBytes(UUID.randomUUID().toString())); //temporary_plate puts.add(put); } // Submit a put request. table.put(puts); LOG.info("Put successfully."); } catch (IOException e) { LOG.error("Put failed ", e); } finally { if (table != null) { try { // Close the HTable object. table.close(); } catch (IOException e) { LOG.error("Close table failed ", e); } } } LOG.info("Exiting testPut."); }

3,从hbase中获取数据

获取数据的方式一共有两种:

(1)用hbase中的唯一键:rowkey,通过get方法获取数据

Get get = new  Get(rowkey) ,通过指定的rowkey来构造Get对象,另外在还可以通过Get对象指定你需要获取的列名get.addColumn(familyName,qulifiers)    

这个方法中需要注意的地方在: 获得cell对象后:(for(Cell cell: results.rawCells()) ) ,从cell中取出数据时,需要将hbase的存储的数据类型Bytes还原成原来的数据类型。  CellUtil.cloneValue(cell)拿到的是Bytes类型的数据,需要通过Bytes.toString() 或Bytes.toInt()等不同的方法转为原来的类型。

例如:插入时原始数据是Int,那么必须调用Bytes.toInt()的方法还原回int,初始数据如果是long类型,必须调用Bytes.toLong()的方法还原回Long, 总结来说就是插入hbase之前是什么样的,出来必须与插入前是相同的数据类型。(否则的话,取出来的数据是乱码的)

@Test
    public void testGet() {
        LOG.info("Entering testGet.");
        // Specify the column family name.
        byte[] familyName = Bytes.toBytes("f1");
        // Specify the column name.
        byte[][] qualifier = { Bytes.toBytes("l1")};
        // Specify RowKey.
        byte[] rowKey = Bytes.toBytes("105f1fd2-7048-4fd3-8c7a-65cf04542be2");
        Table table = null;
        try {
            // Create the Configuration instance.
            table = connection.getTable(TableName.valueOf("table"));
            // Instantiate a Get object.
            Get get = new Get(rowKey);
            // Set the column family name and column name.
            get.addColumn(familyName, qualifier[0]);
            // Submit a get request.
            Result result = table.get(get);
            // Print query results.
            for (Cell cell : result.rawCells()) {
                LOG.info(Bytes.toString(CellUtil.cloneRow(cell)) + ":"
                        + Bytes.toString(CellUtil.cloneFamily(cell)) + ","
                        + Bytes.toString(CellUtil.cloneQualifier(cell)) + ","
                        + Bytes.toString(CellUtil.cloneValue(cell)));
// + Bytes.toLong(CellUtil.cloneValue(cell));
// + Bytes.toInt(CellUtil.cloneValue(cell));
// + Bytes.toBoolean(CellUtil.cloneValue(cell));
} LOG.info(
"Get data successfully."); } catch (IOException e) { LOG.error("Get data failed ", e); } finally { if (table != null) { try { // Close the HTable object. table.close(); } catch (IOException e) { LOG.error("Close table failed ", e); } } } LOG.info("Exiting testGet."); }

 

(2)构建简单的scan对象扫描获取数据

这边也可以构建简单的scan对象,通过表扫描的来获取数据,获得数据的数据类型也是Bytes类型的。关于scan对象我们还可以设置scan的开始rowkey, scan的结束rowkey,scan可以设置缓存大小,可以设置rowkey过滤器,column过滤器等等过滤器。这些会在下一篇章记录。

@Test
    public void testScanData() {
        logger.info("Entering testScanData.");
        Table table = null;
        // Instantiate a ResultScanner object.
        ResultScanner rScanner = null;
        try {
            // Create the Configuration instance.
            table = connection.getTable(TableName.valueOf(TABLENAME));
            // Instantiate a Get object.
            Scan scan = new Scan();
//            scan.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("l1"));
//            scan.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("l2"));
            // Set the cache size.
            scan.setCaching(1000);
            // Submit a scan request.
            rScanner = table.getScanner(scan);
            // Print query results.
            for (Result r = rScanner.next(); r != null; r = rScanner.next()) {
                for (Cell cell : r.rawCells()) {
                    logger.info(Bytes.toString(CellUtil.cloneRow(cell)) + ":"
                            + "\"" + Bytes.toString(CellUtil.cloneQualifier(cell)) + "\":"+"\""
                            + Bytes.toString(CellUtil.cloneValue(cell))+"\"");
                }
            }
            logger.info("Scan data successfully.");
        } catch (IOException e) {
            logger.error("Scan data failed ", e);
        } finally {
            if (rScanner != null) {
                // Close the scanner object.
                rScanner.close();
            }
            if (table != null) {
                try {
                    // Close the HTable object.
                    table.close();
                } catch (IOException e) {
                    logger.error("Close table failed ", e);
                }
            }
        }
        logger.info("Exiting testScanData.");
    }

 




















以上是关于HBase 常用java api获得客户端,创建表,查询,删除的主要内容,如果未能解决你的问题,请参考以下文章

Hbase常用api

HBase编程 API入门之create(管理端而言)

HBase

HBase Shell及其常用命令

hbase java api样例(版本1.3.1,新API)

Java 操作HBase