maven java连接hbase，

和通数据库htsjk.Com2019-09-06 23:19 来源:未知阅读:16896 评论 257 热度2

标签：

maven java连接hbase，

1、pom

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.yj.hbase-test1</groupId>
    <artifactId>hbase-test1</artifactId>
    <version>1.0-SNAPSHOT</version>


    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <spring.version>4.2.2.RELEASE</spring.version>
        <slf4j.version>1.7.7</slf4j.version>
        <xdcs.release.version>0.0.9-SNAPSHOT</xdcs.release.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.11</version>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>jcl-over-slf4j</artifactId>
            <version>${slf4j.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-it</artifactId>
            <version>1.1.2</version>
        </dependency>

    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.6.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>


</project>

2、在resources下新建hbase-site.xml，可以将hbase中conf的这个文件拷贝过来

<?xml version="1.0" encoding="UTF-8"?>

<configuration>
    <!--这里设置让HBase存储文件的地方，可以设置为hdfs://localhost:8000/hbase的hdfs路径-->
    <property>
        <name>hbase.rootdir</name>
        <value>file:///Users/nali/dev_data/hbase/hbase_file</value>
    </property>
    <!--这里设置让HBase存储内建zookeeper文件的地方-->
    <property>
        <name>hbase.zookeeper.property.dataDir</name>
        <value>/Users/nali/dev_data/hbase/hbase_zookeeper</value>
    </property>
    <!--zookeeper地址-->
    <property>
        <name>hbase.zookeeper.quorum</name>
        <value>localhost</value>
    </property>
    <!--zookeeper端口-->
    <property>
        <name>hbase.zookeeper.property.clientPort</name>
        <value>2182</value>
    </property>
    <!--master节点-->hbase.master.info.port
    <property>
        <name>hbase.master</name>
        <value>localhost</value>
    </property>
    <!--master节点端口-->
    <property>
        <name>hbase.master.info.port</name>
        <value>60000</value>
    </property>
</configuration>

3、java代码

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

import org.apache.commons.lang.math.NumberUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.exceptions.DeserializationException;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.spi.LoggerFactory;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;

public class HBaseDemo {

    // 与HBase数据库的连接对象
    Connection connection;

    // 数据库元数据操作对象
    Admin admin;

    Logger logger = org.slf4j.LoggerFactory.getLogger(HBaseDemo.class);

    @Before
    public void setUp() throws Exception {

        // 取得一个数据库连接的配置参数对象
        Configuration conf = HBaseConfiguration.create();

        // 设置连接参数：HBase数据库所在的主机IP
        //conf.set("hbase.zookeeper.quorum", "192.168.137.13");
        // 设置连接参数：HBase数据库使用的端口
        //conf.set("hbase.zookeeper.property.clientPort", "2181");

        // 取得一个数据库连接对象
        connection = ConnectionFactory.createConnection(conf);

        // 取得一个数据库元数据操作对象
        admin = connection.getAdmin();
    }

    /**
     * 创建表
     */
    @Test
    public void createTable() throws IOException{

        System.out.println("---------------创建表 START-----------------");

        // 数据表表名
        String tableNameString = "t_book";

        // 新建一个数据表表名对象
        TableName tableName = TableName.valueOf(tableNameString);

        // 如果需要新建的表已经存在
        if(admin.tableExists(tableName)){

            System.out.println("表已经存在！");
        }
        // 如果需要新建的表不存在
        else{

            // 数据表描述对象
            HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName);

            // 列族描述对象
            HColumnDescriptor family= new HColumnDescriptor("base");;

            // 在数据表中新建一个列族
            hTableDescriptor.addFamily(family);

            // 新建数据表
            admin.createTable(hTableDescriptor);
        }

        System.out.println("---------------创建表 END-----------------");
    }

    /**
     * 查询整表数据
     */
    @Test
    public void queryTable() throws IOException{

        System.out.println("---------------查询整表数据 START-----------------");

        // 取得数据表对象
        Table table = connection.getTable(TableName.valueOf("t_book"));

        // 取得表中所有数据
        ResultScanner scanner = table.getScanner(new Scan());

        // 循环输出表中的数据
        for (Result result : scanner) {

            byte[] row = result.getRow();
            System.out.println("row key is:" + new String(row));

            List<Cell> listCells = result.listCells();
            for (Cell cell : listCells) {

                byte[] familyArray = cell.getFamilyArray();
                byte[] qualifierArray = cell.getQualifierArray();
                byte[] valueArray = cell.getValueArray();

                System.out.println("row value is:" + new String(familyArray) + new String(qualifierArray)
                        + new String(valueArray));
            }
        }

        System.out.println("---------------查询整表数据 END-----------------");

    }

    /**
     * 按行键查询表数据
     */
    @Test
    public void queryTableByRowKey() throws IOException{

        System.out.println("---------------按行键查询表数据 START-----------------");

        // 取得数据表对象
        Table table = connection.getTable(TableName.valueOf("t_book"));

        // 新建一个查询对象作为查询条件
        Get get = new Get("row8".getBytes());

        // 按行键查询数据
        Result result = table.get(get);

        byte[] row = result.getRow();
        System.out.println("row key is:" + new String(row));

        List<Cell> listCells = result.listCells();
        for (Cell cell : listCells) {

            byte[] familyArray = cell.getFamilyArray();
            byte[] qualifierArray = cell.getQualifierArray();
            byte[] valueArray = cell.getValueArray();

            System.out.println("row value is:" + new String(familyArray) + new String(qualifierArray)
                    + new String(valueArray));
        }

        System.out.println("---------------按行键查询表数据 END-----------------");

    }

    /**
     * 按行键查询表数据
     */
    @Test
    public void queryTableByRowKey1() throws IOException{

        System.out.println("---------------按行键查询表数据 START-----------------");

        // 取得数据表对象
        Table table = connection.getTable(TableName.valueOf("t_book"));

        // 新建一个查询对象作为查询条件
        Get get = new Get("row8".getBytes());

        // 按行键查询数据
        Result result = table.get(get);

        byte[] row = result.getRow();
        System.out.println("row key is:" + new String(row));

        List<Cell> listCells = result.listCells();
        for (Cell cell : listCells) {
            System.out.println("family:" +Bytes.toString(CellUtil.cloneFamily(cell)) );
            System.out.println("qualifier:" +Bytes.toString(CellUtil.cloneQualifier(cell)) );
            System.out.println("value:" +Bytes.toString(CellUtil.cloneValue(cell)) );
            System.out.println("-------------------------------");
        }

        System.out.println("---------------按行键查询表数据 END-----------------");

    }

    /**
     * 按条件查询表数据
     */
    @Test
    public void queryTableByCondition() throws IOException{

        System.out.println("---------------按条件查询表数据 START-----------------");

        // 取得数据表对象
        Table table = connection.getTable(TableName.valueOf("t_book"));

        // 创建一个查询过滤器
        Filter filter = new SingleColumnValueFilter(Bytes.toBytes("base"), Bytes.toBytes("name"),
                CompareOp.EQUAL, Bytes.toBytes("bookName6"));

        // 创建一个数据表扫描器
        Scan scan = new Scan();

        // 将查询过滤器加入到数据表扫描器对象
        scan.setFilter(filter);

        // 执行查询操作，并取得查询结果
        ResultScanner scanner = table.getScanner(scan);

        // 循环输出查询结果
        for (Result result : scanner) {
            byte[] row = result.getRow();
            System.out.println("row key is:" + new String(row));

            List<Cell> listCells = result.listCells();
            for (Cell cell : listCells) {
                System.out.println("family:" +Bytes.toString(CellUtil.cloneFamily(cell)) );
                System.out.println("qualifier:" +Bytes.toString(CellUtil.cloneQualifier(cell)) );
                System.out.println("value:" +Bytes.toString(CellUtil.cloneValue(cell)) );
            }
        }

        System.out.println("---------------按条件查询表数据 END-----------------");

    }

    /**
     * 按时间戳条件查询表数据
     */
    @Test
    public void queryTableByCondition1() throws IOException{

        System.out.println("---------------按条件查询表数据 START-----------------");

        // 取得数据表对象
        Table table = connection.getTable(TableName.valueOf("t_book"));

        // 创建一个查询过滤器
        Filter filter = new SingleColumnValueFilter(Bytes.toBytes("base"), Bytes.toBytes("name"),
                CompareOp.EQUAL, Bytes.toBytes("bookName6"));

        // 创建一个数据表扫描器
        Scan scan = new Scan();

        // 将查询过滤器加入到数据表扫描器对象
        scan.setFilter(filter);

        // 执行查询操作，并取得查询结果
        ResultScanner scanner = table.getScanner(scan);

        // 循环输出查询结果
        for (Result result : scanner) {
            byte[] row = result.getRow();
            System.out.println("row key is:" + new String(row));

            List<Cell> listCells = result.listCells();
            for (Cell cell : listCells) {
                System.out.println("family:" +Bytes.toString(CellUtil.cloneFamily(cell)) );
                System.out.println("qualifier:" +Bytes.toString(CellUtil.cloneQualifier(cell)) );
                System.out.println("value:" +Bytes.toString(CellUtil.cloneValue(cell)) );
            }
        }

        System.out.println("---------------按条件查询表数据 END-----------------");

    }

    /**
     * 根据时间戳范围查询
     * @throws IOException
     */
    @Test
    public void getReusltScannerWithSize() throws IOException {
        String startKey = "1503626313145";
        String endKey="1603626313145";
        long size=500;
        // 取得数据表对象
        Table table = connection.getTable(TableName.valueOf("member"));
        ResultScanner rs = null;

        try {
            Scan scan = new Scan();
            scan.setTimeRange(NumberUtils.toLong(startKey), NumberUtils.toLong(endKey));
            scan.setCacheBlocks(false);
            scan.setCaching(500);
            PageFilter filter = new PageFilter(size);
            scan.setFilter(filter);
            rs = table.getScanner(scan);
        } finally {
            if(table != null) {
                table.close();
            }
        }
        System.out.println("start");
        for (Result r : rs) {
            for (Cell cell : r.listCells()) {
                System.out.println("qualifier:" +Bytes.toString(CellUtil.cloneQualifier(cell)) );
                System.out.println("value:" +Bytes.toString(CellUtil.cloneValue(cell)) );
                System.out.println("-------------------------------");
            }
        }
        rs.close();
        System.out.println("end");

    }
    /**
     * 清空表
     */
    @Test
    public void truncateTable() throws IOException{

        System.out.println("---------------清空表 START-----------------");

        // 取得目标数据表的表名对象
        TableName tableName = TableName.valueOf("t_book");

        // 设置表状态为无效
        admin.disableTable(tableName);
        // 清空指定表的数据
        admin.truncateTable(tableName, true);

        System.out.println("---------------清空表 End-----------------");
    }

    /**
     * 删除表
     */
    @Test
    public void deleteTable() throws IOException{

        System.out.println("---------------删除表 START-----------------");

        // 设置表状态为无效
        admin.disableTable(TableName.valueOf("t_book"));
        // 删除指定的数据表
        admin.deleteTable(TableName.valueOf("t_book"));

        System.out.println("---------------删除表 End-----------------");
    }

    /**
     * 删除行
     */
    @Test
    public void deleteByRowKey() throws IOException{

        System.out.println("---------------删除行 START-----------------");

        // 取得待操作的数据表对象
        Table table = connection.getTable(TableName.valueOf("t_book"));

        // 创建删除条件对象
        Delete delete = new Delete(Bytes.toBytes("row2"));

        // 执行删除操作
        table.delete(delete);

        System.out.println("---------------删除行 End-----------------");

    }

    /**
     * 删除行（按条件）
     */
    @Test
    public void deleteByCondition() throws IOException, DeserializationException{

        System.out.println("---------------删除行（按条件） START-----------------");

        // 步骤1：调用queryTableByCondition()方法取得需要删除的数据列表

        // 步骤2：循环步骤1的查询结果，对每个结果调用deleteByRowKey()方法

        System.out.println("---------------删除行（按条件） End-----------------");

    }

    /**
     * 新建列族
     */
    @Test
    public void addColumnFamily() throws IOException{

        System.out.println("---------------新建列族 START-----------------");

        // 取得目标数据表的表名对象
        TableName tableName = TableName.valueOf("t_book");

        // 创建列族对象
        HColumnDescriptor columnDescriptor = new HColumnDescriptor("more");

        // 将新创建的列族添加到指定的数据表
        admin.addColumn(tableName, columnDescriptor);

        System.out.println("---------------新建列族 END-----------------");
    }

    /**
     * 删除列族
     */
    @Test
    public void deleteColumnFamily() throws IOException{

        System.out.println("---------------删除列族 START-----------------");

        // 取得目标数据表的表名对象
        TableName tableName = TableName.valueOf("t_book");

        // 删除指定数据表中的指定列族
        admin.deleteColumn(tableName, "more".getBytes());

        System.out.println("---------------删除列族 END-----------------");
    }

    /**
     * 插入数据
     */
    @Test
    public void insert() throws IOException{

        System.out.println("---------------插入数据 START-----------------");

        // 取得一个数据表对象
        Table table = connection.getTable(TableName.valueOf("t_book"));

        // 需要插入数据库的数据集合
        List<Put> putList = new ArrayList<Put>();

        Put put;

        // 生成数据集合
        for(int i = 0; i < 10; i++){
            put = new Put(Bytes.toBytes("row" + i));
            put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("name"), Bytes.toBytes("bookName" + i));
            put.addColumn(Bytes.toBytes("base"), Bytes.toBytes("age"), Bytes.toBytes("1" + i));

            putList.add(put);
        }

        // 将数据集合插入到数据库
        table.put(putList);

        System.out.println("---------------插入数据 END-----------------");
    }

}