Hbase安装以及基本操作

Hbase基礎學習

文章目录

• Hbase基礎學習
• 一，HBASE的安裝（MAC環境）
• • 1.1 修改配置文件
  • • 1.1.1 hbase-env.sh
    • 1.1.2 hbase-site.xml
  • 1.2 配置環境變量
  • 1.3 運行
  • 1.4 常見問題
  • • 1.4.1 JAVA_HOME is not set
    • 1.4.2 KeeperErrorCode = ConnectionLoss for /hbase
    • 1.4.3 hadoop安裝後hbase的ruby出錯
• 二，Hbase常用命令
• 三，Java使用Hbase
• • 3.1 过滤器Filter
  • 3.2 Hbase協處理器
  • • 3.2.1 Observer Coprocessor 的类型
    • 3.2.2 配置協處理器

一，HBASE的安裝（MAC環境）

下載路徑： http://archive.apache.org/dist/hbase/ (2.4.0)

下載完解壓到 /usr/local/ 目錄下的 hbase-2.4.0 目錄下

tar -zxvf /Data/hbase-2.4.0-bin.tar.gz /Data/hbase-2.4.0/

1.1 修改配置文件

進入 hbase-2.4.0/conf目錄，修改hbase-env.sh和hbase-site.xml文件

1.1.1 hbase-env.sh

在該文件下配置java路徑（JAVA_HOME）,hbase使用HBASE自帶的zookeeper（HBASE_MANAGES_ZK）,和pid存儲位置（HBASE_PID_DIR）

export JAVA_HOME=/Library/Java/JavaVirtualMachines/jdk1.8.0_121.jdk/Contents/Home/
export HBASE_MANAGES_ZK=true
export HBASE_PID_DIR=/usr/local/hbase-2.4.0/pids

1.1.2 hbase-site.xml

在該文件的configuration標籤中加入以下配置：

 

    hbase.cluster.distributed
    false
  
  
  
    hbase.tmp.dir
    ./tmp
   
  
  
    hbase.unsafe.stream.capability.enforce
    false
  
  
  
    hbase.rootdir
     
    hdfs://10.197.29.203:9000/hbase-tmp
     
   
  

1.2 配置環境變量

修改.bashrc文件

vim ~/.bashrc

 export PATH=$PATH:/usr/local/hbase-2.4.0/bin

source ~/.bashrc

查看HBASE是否安裝成功

/usr/local/hbase-2.4.0/bin/hbase version

1.3 運行

啟動Hbase，操作HBASE數據庫

bin/start-hbase.sh
bin/hbase shell

jps查看啟動情況

jps

此時應該存在一個名稱為 HMaster 的進程

關閉Hbase

bin/stop-hbase.sh

1.4 常見問題 1.4.1 JAVA_HOME is not set

进入/usr/local/hbase-2.4.0目录,修改conf/目录下的hbase-env.sh文件,找到 export JAVA_HOME  这一行注释内容,将原来的java环境删除,换上自己的java路径就可以了

5. 啟動Hbase，操作HBASE數據庫

bin/start-hbase.sh
bin/hbase shell

6. 關閉Hbase

bin/stop-hbase.sh

1.4.2 KeeperErrorCode = ConnectionLoss for /hbase

啟動/關閉hbase等命令時不要使用 sudo 命令

1.4.3 hadoop安裝後hbase的ruby出錯

LoadError: load error: irb/completion -- java.lang.IncompatibleClassChangeError

將lib下的jruby-complete-x.x.x.jar包修改版本，（我的hbase是2.4.0，修改jar包成jruby-complete-9.1.17.0.jar）

然後在hadoop的 hadoop-2.4.0/share/hadoop/yarn/lib目錄下，也修改 jline-x.x.x.jar這個jar包的版本（我的hadoop是2.4.0，修改jar包成 jline-3.9.0.jar）

二，Hbase常用命令

命名	描述	语法
whoami	我是谁	whoami
version	返回hbase集群的状态信息	version
status	返回hbase集群的状态信息	status
table_help	查看如何操作表	table_help
create	创建表	create ‘表名’, ‘列族1’ create ‘表名’, ‘列族1’, ‘列族2’, ‘列族N’
alter	修改列族	添加一个列族：alter ‘表名’, 列族X’ 删除列族：alter ‘表名’, {NAME=> ‘列族’, METHOD=> ‘delete’}
describe	显示表相关的详细信息	describe/desc ‘表名’
list	列出hbase中存在的所有表	list
exists	测试表是否存在	exists ‘表名’
put	添加或修改的表的值	put ‘表名’, ‘行键’, ‘列族名’, ‘列值’ put ‘表名’, ‘行键’, ‘列族名:列名’, ‘列值’
scan	通过对表的扫描来获取对用的值	scan ‘表名’ 扫描某个列族： scan ‘表名’, {COLUMN=>‘列族名’} 扫描某个列族的某个列： scan ‘表名’, {COLUMN=>‘列族名:列名’} 查询同一个列族的多个列： scan ‘表名’, {COLUMNS => [ ‘列族名1:列名1’, ‘列族名1:列名2’, …]}
get	获取行或单元（cell）的值	get ‘表名’, ‘行键’ get ‘表名’, ‘行键’, ‘列族名’
count	统计表中行的数量	count ‘表名’
incr	增加指定表行或列的值	incr ‘表名’, ‘行键’, ‘列族:列名’, 步长值
get_counter	获取计数器	get_counter ‘表名’, ‘行键’, ‘列族:列名’
delete	删除指定对象的值（可以为表，行，列对应的值，另外也可以指定时间戳的值））	删除列族的某个列： delete ‘表名’, ‘行键’, ‘列族名:列名’
deleteall	删除指定行的所有元素值	deleteall ‘表名’, ‘行键’
enable	使表有效	enable ‘表名’
disable	使表无效	disable ‘表名’
is_enabled	是否启用	is_enabled ‘表名’
is_disabled	是否无效	is_disabled ‘表名’
drop	删除表	drop的表必须是disable的 disable ‘表名’ drop ‘表名’
truncate	清空表，重新创建指定表	truncate ‘表名’
shutdown	关闭hbase集群（与exit不同）
tools	列出hbase所支持的工具
exit	退出

scan參數：

名稱	描述	使用
COLUMN	單個列族名稱	{COLUMN=>‘XXX’} 或 {COLUMN=>‘XXX:xxx’}
COLUMNS	多個列族名稱	{COLUMNS=>[‘XXX’, ‘YYY:yyy’ …]}
LIMIT	限制查找條數	{LIMIT=>1}
TIMERANGE	限制時間範圍	{TIMERANGE=>[startTimeLong,endTimeLong]}
FILTER	過濾部分
PrefixFilter	rowkey前綴過濾	scan ‘user’,{FILTER=>“PrefixFilter(‘key-prefix’)”}
RowFilter	行主鍵過濾	scan ‘user’,{FILTER=>“RowFilter(=,‘substring:key001’)”} scan ‘user’,{FILTER=>“RowFilter(>=,‘binary:a’)”}
KeyOnlyFilter	只对单元格的键进行过滤和显示，不显示值	scan ‘user’,{FILTER => “KeyOnlyFilter()”}
FirstKeyOnlyFilter	只扫描显示相同键的第一个单元格，其键值对会显示出来（用來計數效率很高）	scan ‘user’,{FILTER => “FirstKeyOnlyFilter()”}
FamilyFilter	列族過濾	scan ‘user’,FILTER=>“FamilyFilter(=,‘substring:base’)” scan ‘user’,FILTER=>“FamilyFilter(>,‘binary:base’)”
QualifierFilter	列名稱過濾	scan ‘user’,FILTER=>“QualifierFilter(=,‘substring:name’)” scan ‘user’,FILTER=>“QualifierFilter(<=,‘binary:na’)”
ColumnPrefixFilter	对列名称的前缀进行过滤	scan ‘user’, FILTER=>“ColumnPrefixFilter(‘na’)”
MultipleColumnPrefixFilter	可以指定多个前缀对列名称过滤	scan ‘user’,FILTER=>“MultipleColumnPrefixFilter(‘name’, ‘math’)”
ValueFilter	值過濾	scan ‘user’, FILTER=>“ValueFilter(=,‘substring:test’)” get ‘user’, ‘row-001’, FILTER=>“ValueFilter(=,‘substring:test’)”
SingleColumnValueFilter	在指定的列族和列中进行比较的值过滤器	scan ‘user’, FILTER=>“SingleColumnValueFilter(‘base’,‘name’,=,‘substring:test’)”
SingleColumnValueExcludeFilter	排除匹配成功的值
ColumnCountGetFilter	限制每个逻辑行返回键值对的个数	scan ‘user’, FILTER=>“ColumnCountGetFilter(2)”
PageFilter	对显示结果按行进行分页显示（一頁就是一個行的數據）	scan ‘user’, {STARTROW=>‘key001’,ENDROW=>‘key005’, FILTER=>“PageFilter(1)”}
ColumnPaginationFilter	ColumnPaginationFilter(limit, offset)，对一行的所有列分页，只返回[offset,offset+limit] 范围内的列	scan “user”, {STARTROW=>‘’, FILTER=>“ColumnPaginationFilter(2, 0)”}

三，Java使用Hbase 3.1 过滤器Filter

第一个参数有很多种取值以匹配多种场景，取值表格如下：

hbase命令	java操作	描述
<	CompareOperator.LESS	匹配小宇设定值的值
	CompareOperator.LESS_OR_EQUAL	匹配小宇或等于设定值的值
	CompareOperator.EQUAL	匹配等于设定值的值
	CompareOperator.NOT_EQUAL	匹配与设定值不相等的值
	CompareOperator.GREATER_OR_EQUAL	匹配大于或等于设定值的值
	CompareOperator.GREATER	匹配大于设定值的值
	CompareOperator.NO_OP	排除一切值

比较器有很多子类，他们的功能如下：

比较器	描述
BinaryComparator	使用Bytes.compareTo()比较当前值与阈值
BinaryPrefixComparator	与上面类似，但是是从左端开始前缀匹配
NullComparator	不做匹配，只判断当前值是不是null
BitComparator	通过BitwiseOp类提供的按位与（AND）、或（OR）、异或（XOR）操作执行位级比较
RegexStringComparator	根据一个正则表达式，在实例化这个比较器的时候去匹配表中的数据
SubStringComparator	把阈值和表中数据当做String实例，同时通过contains()操作匹配字符串

注意：

• 最后三种比较器，即BitComparator、RegexStringComparator 和 SubStringComparator，只能与EQUAL和NOT_EQUAL运算符一起使用，因为这些比较器compareTo()方法匹配的时候返回结果是0或者1，如果和GREATER或LESS运算符搭配使用，就会产生错误的结果。

• 使用SingleColumnValueFilter的前提是查詢的COLUMN或COLUMNS中，必須包含此列。

  SingleColumnValueFilter(列族，列名， 比較符，比較器【是否忽略不存在此列的記錄（默認false）， 是否只獲取最新版本的記錄（默認true）】)

• 若同時使用多個過濾器，需要使用類FilterList。對於AND操作則傳入參數FilterList.Operator.MUST_PASS_ALL，OR操作則傳入參數FilterList.Operator.MUST_PASS_ONE

//定義過濾器集
FilterList allFilter = new FilterList(FilterList.Operator.MUST_PASS_ALL);
FilterList orFilter = new FilterList(FilterList.Operator.MUST_PASS_ONE);
//定義過濾器
SingleColumnValueFilter filter1 = new SingleColumnValueFilter(x,x,x,x);
SingleColumnValueFilter filter2 = new SingleColumnValueFilter(y,y,y,y);
FirstKeyOnlyFilter firstKeyOnlyFilter = new FirstKeyOnlyFilter();
//條件
orFilter.addFilter(filter1);
orFilter.addFilter(filter2);
allFilter.addFilter(orFilter);
allFilter.addFilter(firstKeyOnlyFilter);

等同於

FILTER=>"( SingleColumnValueFilter(x,x,x,x) OR SingleColumnValueFilter(y,y,y,y) ) AND FirstKeyOnlyFilter()"

			//RowKey 过滤器
            System.out.println("===============================杭键 过滤器================================");
            System.out.println("==================RowKey 过滤器===================");
            Filter filter = new RowFilter(CompareOperator.EQUAL, new SubstringComparator("003"));
            System.out.println(scan(connection, "user", null, null, filter));
            filter = new RowFilter(CompareOperator.LESS, new BinaryComparator(Bytes.toBytes("key003")));
            System.out.println(scan(connection, "user", null, null, filter));

            System.out.println("==================PrefixFilter 过滤器===================");
            filter = new PrefixFilter(Bytes.toBytes("row"));
            System.out.println(scan(connection, "user", null, null, filter));

            System.out.println("==================FirstKeyOnlyFilter 过滤器===================");
            filter = new FirstKeyOnlyFilter();
            System.out.println(scan(connection, "user", null, null, filter));

            System.out.println("==================InclusiveStopFilter 过滤器===================");
            filter = new InclusiveStopFilter(Bytes.toBytes("key006"));
            System.out.println(scan(connection, "user", "key003", null, filter));

            System.out.println("===============================列 过滤器================================");
            System.out.println("==================FamilyFilter 过滤器===================");
            filter = new FamilyFilter(CompareOperator.EQUAL, new SubstringComparator("score"));
            System.out.println(scan(connection, "user", null, null, filter));

            System.out.println("==================QualifierFilter 过滤器===================");
            filter = new QualifierFilter(CompareOperator.EQUAL, new SubstringComparator("math"));
            System.out.println(scan(connection, "user", null, null, filter));

            System.out.println("==================ColumnPrefixFilter 过滤器===================");
            filter = new ColumnPrefixFilter(Bytes.toBytes("age"));
            System.out.println(scan(connection, "user", null, null, filter));

            System.out.println("==================MultipleColumnPrefixFilter 过滤器===================");
            filter = new MultipleColumnPrefixFilter(Bytes.toArray(Arrays.asList(Bytes.toBytes("age"), Bytes.toBytes("math"))));
            System.out.println(scan(connection, "user", null, null, filter));

            System.out.println("===============================值 过滤器================================");
            System.out.println("==================ValueFilter 过滤器===================");
            filter = new ValueFilter(CompareOperator.GREATER, new BinaryComparator(Bytes.toBytes("test06")));
            System.out.println(scan(connection, "user", null, null, filter));

            System.out.println("==================SingleColumnValueFilter 过滤器===================");
            filter = new SingleColumnValueFilter(Bytes.toBytes("base"), Bytes.toBytes("name"), CompareOperator.LESS, new BinaryComparator(Bytes.toBytes("test05")));
            System.out.println(scan(connection, "user", null, null, filter));

            System.out.println("===============================其它 过滤器================================");
            System.out.println("==================PageFilter 过滤器===================");
            filter = new PageFilter(2);
            System.out.println(scan(connection, "user", null, null, filter));

            System.out.println("==================ColumnPaginationFilter 过滤器===================");
            filter = new ColumnPaginationFilter(2, 1);
            System.out.println(scan(connection, "user", "key001", "key002", filter));

注意： 使用過濾器鏈FilterList的時候，要按照過濾的順序進行添加。

3.2 Hbase協處理器 3.2.1 Observer Coprocessor 的类型

RegionObserver

一个 RegionObserver 协处理器允许您观察一个区域上的事件，比如 Get 和 Put 操作。看到RegionObserver。

RegionServerObserver

RegionServerObserver 允许您观察与 RegionServer 操作相关的事件，例如启动、停止或执行合并、提交或回滚。看到RegionServerObserver。

MasterObserver

MasterObserver 允许您观察与 HBase 主节点相关的事件，比如表创建、删除或模式修改。看到MasterObserver。

WalObserver

WalObserver 允许您观察与写入提前写入日志(WAL)相关的事件。看到WALObserver。

3.2.2 配置協處理器

加載協處理器的兩種方式

1. 配置文件加載（靜態配置）

   启动全局aggregation，能过操纵所有的表上的数据。通过修改hbase-site.xml这个文件。

   RegionObservers

    
       hbase.coprocessor.region.classes 	                        org.apache.hadoop.hbase.coprocessor.AggregateImplementation 
   
   

   WALObservers

   
       hbase.coprocessor.wal.classes
       com.lmc.aggregation.MyWALObservers 
   
   

   MasterObservers

   
       hbase.coprocessor.master.classes
       com.lmc.aggregation.MyMasterObservers 
   
   

   如果想同时配置多个协处理器，可以用逗号分隔多个协处理器的类名。

2. shell加載（動態配置）

   启用表aggregation，只对特定的表生效。通过HBase Shell 来实现

   步驟：

   （1）disable指定表；（disable ‘test’）

   （2）添加aggregation；（alter ‘test’, METHOD => ‘table_att’,‘coprocessor’=>‘/usr/local/hbase-2.4.0/lib/lmc/lmc-hbase-region-eggregate.jar|com.lmc.aggregation.MyRegionObserver1||’）

   （3）重启指定表。（enable ‘test’）

   其他：刪除aggregation（alter ‘test’, METHOD => ‘table_att_unset’,NAME=>‘coprocessor$1’）

​ （4）重啟HBASE服務

alter ‘test’, METHOD => ‘table_att’,‘coprocessor’=>‘①|②|③|④’

參數介紹：

①：协处理器Jar包路径，要保证所有的RegionServer可读取到。也可以是本地路径，不过建议是放在HDFS上。

②：协处理器的完整类名。

③：协处理器优先级，整数表示。可以为空。

④：传递给协处理器的参数，可以为空。

MyRegionObserver1.java

package com.lmc.aggregation;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
import org.apache.hadoop.hbase.coprocessor.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.wal.WALEdit;

import java.io.IOException;
import java.util.List;
import java.util.Optional;
import java.util.logging.Logger;


public class MyRegionObserver1 implements RegionObserver, RegionCoprocessor {

    private static final Logger LOG = Logger.getLogger(String.valueOf(MyRegionObserver1.class));

    public static Configuration configuration = null;

    private static Table table = null;

    static {
        configuration = HBaseConfiguration.create();
        // 设置配置参数    localhost寫成IP地址時報錯
        configuration.set("hbase.zookeeper.quorum", "localhost");
        configuration.set("hbase.zookeeper.property.clientPort", "2181");
        try {
            Connection connection = getConnection();
            table = connection.getTable(TableName.valueOf("test1"));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void postPut(ObserverContext c, Put put, WALEdit edit, Durability durability) throws IOException {
        LOG.info("================start to postPut=============");
        //根據rowKey生成put語句
        Put put_0 = new Put(put.getRow());
        put_0.addColumn(Bytes.toBytes("base"), Bytes.toBytes("post"), Bytes.toBytes(("xxx")));
        table.put(put_0);
        table.close();

    }

    @Override
    public void prePut(ObserverContext c, Put put, WALEdit edit, Durability durability) throws IOException {
        LOG.info("================start to prePut=============");
        //根據rowKey生成put語句
        Put put_0 = new Put(put.getRow());
        put_0.addColumn(Bytes.toBytes("base"), Bytes.toBytes("pre"), Bytes.toBytes(("xxx")));
        table.put(put_0);
        table.close();
    }

    @Override
    public void preGetOp(ObserverContext c, Get get, List result) {
        LOG.info("================start to preGetOp=============");
    }

    @Override
    public void postGetOp(ObserverContext c, Get get, List result) {
        LOG.info("================start to postGetOp=============");
    }

    @Override
    public Optional getRegionObserver() {
        return Optional.of(this);
    }

    
    public static Connection getConnection() throws IOException {

        return ConnectionFactory.createConnection(configuration);
    }

    
    public static void close(Connection _zConn) {
        try {
            if (_zConn != null) {
                _zConn.close();
            }
        } catch (IOException e) {
            System.out.println("连接关闭失败！");
            e.printStackTrace();
        }
    }

//    public static void main(String[] args) throws IOException {
//        Put put_0 = new Put(Bytes.toBytes("key001"));
//        put_0.addColumn(Bytes.toBytes("base"), Bytes.toBytes("pre"), Bytes.toBytes(("xxx")));
//        table.put(put_0);
//        table.close();
//    }

}