HBase 1.2.5 Java API教程

作者: FX_SKY | 来源:发表于2017-05-25 18:20 被阅读5856次

关于HBase 1.2 安装&使用参考：HBase Reference Guide。本文重点讲解 HBase 1.2.5 Java API使用。

HBase API

HBase 1.2 API 常用类：

使用org.apache.hadoop.hbase.client.ConnectionFactory来创建HBase数据库连接org.apache.hadoop.hbase.client.Connection；
表名使用特定的类org.apache.hadoop.hbase.TableName 而不是字符串；
使用Admin 新建、删除表，可通过org.apache.hadoop.hbase.client.Connection的getAdmin()方法获取；
使用Table 定义表，可通过org.apache.hadoop.hbase.client.Connection的getTable()方法获取；
使用Put、Get、Scan实现对表的put、get、scan操作。

1.maven依赖

    <!--HBase-->
    <dependency>
      <groupId>org.apache.hbase</groupId>
      <artifactId>hbase-client</artifactId>
      <version>1.2.5</version>
    </dependency>

2.create 操作

Admin类的createTable方法，如下：

    public void createTable(Connection connection, TableName tableName, String... columnFamilies) throws IOException {
        Admin admin = null;
        try {
            admin = connection.getAdmin();
            if(admin.tableExists(tableName)){
                logger.warn("table:{} exists!", tableName.getName());
            }else{
                HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);
                for(String columnFamily : columnFamilies) {
                    tableDescriptor.addFamily(new HColumnDescriptor(columnFamily));
                }
                admin.createTable(tableDescriptor);
                logger.info("create table:{} success!", tableName.getName());
            }
        } finally {
            if(admin!=null) {
                admin.close();
            }
        }
    }

3. put操作

通过 org.apache.hadoop.hbase.client.Put 来操作，如下：

    /**批量插入可以使用 Table.put(List<Put> list)**/
    public void put(Connection connection, TableName tableName,
                    String rowKey, String columnFamily, String column, String data) throws IOException {

        Table table = null;
        try {
            table = connection.getTable(tableName);
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(data));
            table.put(put);
        } finally {
            if(table!=null) {
                table.close();
            }
        }
    }

4. get操作

    //根据row key获取表中的该行数据
    public void get(Connection connection,TableName tableName,String rowKey) throws IOException {
        Table table = null;
        try {
            table = connection.getTable(tableName);
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = table.get(get);
            NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> navigableMap = result.getMap();
            for(Map.Entry<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> entry : navigableMap.entrySet()){

                logger.info("columnFamily:{}", Bytes.toString(entry.getKey()));
                NavigableMap<byte[], NavigableMap<Long, byte[]>> map =entry.getValue();
                for(Map.Entry<byte[], NavigableMap<Long, byte[]>> en:map.entrySet()){
                    System.out.print(Bytes.toString(en.getKey())+"##");
                    NavigableMap<Long, byte[]> nm = en.getValue();
                    for(Map.Entry<Long, byte[]> me : nm.entrySet()){
                        logger.info("column key:{}, value:{}", me.getKey(), me.getValue());
                    }
                }
            }
        } finally {
            if(table!=null) {
                table.close();
            }
        }
    }

5. scan操作


    public void scan(Connection connection, TableName tableName) throws IOException {
        Table table = null;
        try {
            table = connection.getTable(tableName);
            ResultScanner rs = null;
            try {
                //Scan scan = new Scan(Bytes.toBytes("u120000"), Bytes.toBytes("u200000"));
                rs = table.getScanner(new Scan());
                for(Result r:rs){
                    NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> navigableMap = r.getMap();
                    for(Map.Entry<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> entry : navigableMap.entrySet()){
                        logger.info("row:{} key:{}", Bytes.toString(r.getRow()), Bytes.toString(entry.getKey()));
                        NavigableMap<byte[], NavigableMap<Long, byte[]>> map =entry.getValue();
                        for(Map.Entry<byte[], NavigableMap<Long, byte[]>> en:map.entrySet()){
                            System.out.print(Bytes.toString(en.getKey())+"##");
                            NavigableMap<Long, byte[]> ma = en.getValue();
                            for(Map.Entry<Long, byte[]>e: ma.entrySet()){
                                System.out.print(e.getKey()+"###");
                                System.out.println(Bytes.toString(e.getValue()));
                            }
                        }
                    }
                }
            } finally {
                if(rs!=null) {
                    rs.close();
                }
            }
        } finally {
            if(table!=null) {
                table.close();
            }
        }
    }

6. delete操作


    //删除表中的数据
    public void deleteTable(Connection connection, TableName tableName) throws IOException {
        Admin admin = null;
        try {
            admin = connection.getAdmin();
            if (admin.tableExists(tableName)) {
                //必须先disable, 再delete
                admin.disableTable(tableName);
                admin.deleteTable(tableName);
            }
        } finally {
            if(admin!=null) {
                admin.close();
            }
        }
    }

测试


    public void testCrud() {
        Connection connection = null;
        try {
            connection = HBaseConnectionUtils.getConnection();
            TableName tableName = TableName.valueOf("demo");

            //创建HBase表
            createTable(connection, tableName, "cf1", "cf2");

            //put
            String rowKey = "u12000";
            put(connection, tableName, rowKey, "cf1", "name", "ricky");
            put(connection, tableName, rowKey, "cf1", "password", "root");
            put(connection, tableName, rowKey, "cf1", "age", "28");

            //get
            get(connection, tableName, rowKey);

            //scan
            scan(connection, tableName);

            //delete
            deleteTable(connection, tableName);

        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

完整示例代码

package com.mindflow.hbase.tutorials.crud;

import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
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.Map;
import java.util.NavigableMap;

/**
 * http://hbase.apache.org/1.2/apidocs/index.html
 *
 * @author Ricky Fung
 */
public class HBaseCrudDemo {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    public static void main(String[] args) {

        new HBaseCrudDemo().testCrud();
    }

    public void testCrud() {
        Connection connection = null;
        try {
            connection = HBaseConnectionUtils.getConnection();
            TableName tableName = TableName.valueOf("demo");

            //创建HBase表
            createTable(connection, tableName, "cf1", "cf2");

            //put
            String rowKey = "u12000";
            put(connection, tableName, rowKey, "cf1", "name", "ricky");
            put(connection, tableName, rowKey, "cf1", "password", "root");
            put(connection, tableName, rowKey, "cf1", "age", "28");

            //get
            get(connection, tableName, rowKey);

            //scan
            scan(connection, tableName);

            //delete
            deleteTable(connection, tableName);

        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public void scan(Connection connection, TableName tableName) throws IOException {
        Table table = null;
        try {
            table = connection.getTable(tableName);
            ResultScanner rs = null;
            try {
                //Scan scan = new Scan(Bytes.toBytes("u120000"), Bytes.toBytes("u200000"));
                rs = table.getScanner(new Scan());
                for(Result r:rs){
                    NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> navigableMap = r.getMap();
                    for(Map.Entry<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> entry : navigableMap.entrySet()){
                        logger.info("row:{} key:{}", Bytes.toString(r.getRow()), Bytes.toString(entry.getKey()));
                        NavigableMap<byte[], NavigableMap<Long, byte[]>> map =entry.getValue();
                        for(Map.Entry<byte[], NavigableMap<Long, byte[]>> en:map.entrySet()){
                            System.out.print(Bytes.toString(en.getKey())+"##");
                            NavigableMap<Long, byte[]> ma = en.getValue();
                            for(Map.Entry<Long, byte[]>e: ma.entrySet()){
                                System.out.print(e.getKey()+"###");
                                System.out.println(Bytes.toString(e.getValue()));
                            }
                        }
                    }
                }
            } finally {
                if(rs!=null) {
                    rs.close();
                }
            }
        } finally {
            if(table!=null) {
                table.close();
            }
        }
    }

    //根据row key获取表中的该行数据
    public void get(Connection connection,TableName tableName,String rowKey) throws IOException {
        Table table = null;
        try {
            table = connection.getTable(tableName);
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = table.get(get);
            NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> navigableMap = result.getMap();
            for(Map.Entry<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> entry : navigableMap.entrySet()){

                logger.info("columnFamily:{}", Bytes.toString(entry.getKey()));
                NavigableMap<byte[], NavigableMap<Long, byte[]>> map =entry.getValue();
                for(Map.Entry<byte[], NavigableMap<Long, byte[]>> en:map.entrySet()){
                    System.out.print(Bytes.toString(en.getKey())+"##");
                    NavigableMap<Long, byte[]> nm = en.getValue();
                    for(Map.Entry<Long, byte[]> me : nm.entrySet()){
                        logger.info("column key:{}, value:{}", me.getKey(), me.getValue());
                    }
                }
            }
        } finally {
            if(table!=null) {
                table.close();
            }
        }
    }

    /**批量插入可以使用 Table.put(List<Put> list)**/
    public void put(Connection connection, TableName tableName,
                    String rowKey, String columnFamily, String column, String data) throws IOException {

        Table table = null;
        try {
            table = connection.getTable(tableName);
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(data));
            table.put(put);
        } finally {
            if(table!=null) {
                table.close();
            }
        }
    }

    public void createTable(Connection connection, TableName tableName, String... columnFamilies) throws IOException {
        Admin admin = null;
        try {
            admin = connection.getAdmin();
            if(admin.tableExists(tableName)){
                logger.warn("table:{} exists!", tableName.getName());
            }else{
                HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);
                for(String columnFamily : columnFamilies) {
                    tableDescriptor.addFamily(new HColumnDescriptor(columnFamily));
                }
                admin.createTable(tableDescriptor);
                logger.info("create table:{} success!", tableName.getName());
            }
        } finally {
            if(admin!=null) {
                admin.close();
            }
        }
    }

    //删除表中的数据
    public void deleteTable(Connection connection, TableName tableName) throws IOException {
        Admin admin = null;
        try {
            admin = connection.getAdmin();
            if (admin.tableExists(tableName)) {
                //必须先disable, 再delete
                admin.disableTable(tableName);
                admin.deleteTable(tableName);
            }
        } finally {
            if(admin!=null) {
                admin.close();
            }
        }
    }

    public void disableTable(Connection connection, TableName tableName) throws IOException {
        Admin admin = null;
        try {
            admin = connection.getAdmin();
            if(admin.tableExists(tableName)){
                admin.disableTable(tableName);
            }
        } finally {
            if(admin!=null) {
                admin.close();
            }
        }
    }
}

参考资料

HBase 1.2 Documentation

源码下载

https://github.com/TiFG/hbase-tutorials/tree/master/hbase-crud-demo

HBase 1.2.5 Java API教程

HBase API

1.maven依赖

2.create 操作

3. put操作

4. get操作

5. scan操作

6. delete操作

测试

完整示例代码

参考资料

源码下载

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

我爱编程