分布式数据库系统——HBase_hbase数据库

#指定regionserver节点
linux121
linux122
linux123

linux122

export HBASE_HOME=/opt/lagou/servers/hbase-1.3.1
export PATH=$PATH:$HBASE_HOME/bin

rsync-script hbase-1.3.1
rsync-script /etc/profile

hbase shell

hbase(main):001:0> help

hbase(main):006:0> list

hbase(main):001:0> create 'lagou', 'base_info', 'extra_info'

create 'lagou', {NAME => 'base_info', VERSIONS => '3'},{NAME => 'extra_info',VERSIONS => '3'}

hbase(main):001:0> put 'lagou', 'rk1', 'base_info:name', 'wang'

hbase(main):001:0> put 'lagou', 'rk1', 'base_info:age', 30

hbase(main):001:0> put 'lagou', 'rk1', 'extra_info:address', 'shanghai'

hbase(main):001:0> get 'lagou', 'rk1'

hbase(main):001:0> get 'lagou', 'rk1', 'base_info'

hbase(main):008:0> get 'lagou', 'rk1', 'base_info:name', 'base_info:age'

hbase(main):010:0> get 'lagou', 'rk1', 'base_info', 'extra_info'

hbase(main):011:0> get 'lagou', 'rk1', {COLUMN => ['base_info', 'extra_info']}

hbase(main):012:0> get 'lagou', 'rk1', {COLUMN => ['base_info:name', 'extra_info:address']}

hbase(main):001:0> get 'lagou', 'rk1', {FILTER => "ValueFilter(=, 'binary:wang')"}

hbase(main):001:0> get 'lagou', 'rk1', {FILTER => "(QualifierFilter(=,'substring:a'))"}

hbase(main):000:0> scan 'lagou'

hbase(main):001:0> scan 'lagou', {COLUMNS => 'base_info'}
hbase(main):002:0> scan 'lagou', {COLUMNS => 'base_info', RAW => true, VERSIONS => 3}
## scan时可以设置是否开启Raw模式,开启Raw模式会返回包括已添加删除标记但是未实际删除的数据
## VERSIONS指定查询的最大版本数

hbase(main):001:0> scan 'lagou', {COLUMNS => ['base_info', 'extra_info'], FILTER => " (QualifierFilter(=,'substring:a'))"}

hbase(main):001:0> scan 'lagou', {COLUMNS => 'base_info', STARTROW => 'rk1', ENDROW => 'rk3'}

hbase(main):001:0> scan 'lagou',{FILTER=>"PrefixFilter('rk')"}

hbase(main):030:0> put 'lagou', 'rk1', 'base_info:name', 'liang'

hbase(main):002:0> delete 'lagou', 'rk1', 'base_info:name'

hbase(main):033:0> delete 'lagou', 'rk1', 'base_info:age'

hbase(main):035:0> alter 'lagou', 'delete' => 'base_info'

hbase(main):001:0> truncate 'lagou'

#先disable 再drop
hbase(main):036:0> disable 'lagou'
hbase(main):037:0> drop 'lagou'
#如果不进行disable，直接drop会报错
ERROR: Table user is enabled. Disable it first.

<property>
    <name>hbase.hregion.memstore.flush.size</name>
    <value>134217728</value>
</property>

<property>
    <name>hbase.regionserver.optionalcacheflushinterval</name>
    <value>3600000</value>
</property>

<property>
    <name>hbase.regionserver.global.memstore.size</name>
    <value>0.4</value>
</property>

flush tableName

<!--待合并⽂件数据必须大于等于下⾯这个值-->
<property>
    <name>hbase.hstore.compaction.min</name>
    <value>3</value>
</property>
<!--待合并文件数据必须⼩于等于下⾯这个值-->
<property>
    <name>hbase.hstore.compaction.max</name>
    <value>10</value>
</property>
<!--默认值为128m,表示文件⼤⼩小于该值的store file一定会加⼊到minor compaction的store file中 -->
<property>
    <name>hbase.hstore.compaction.min.size</name>
    <value>134217728</value>
</property>
<!--默认值为LONG.MAX_VALUE，表示⽂件⼤小大于该值的store file一定会被minor compaction排除-->
<property>
    <name>hbase.hstore.compaction.max.size</name>
    <value>9223372036854775807</value> 
</property>

<!--默认值为7天进⾏一次⼤合并，-->
<property>
    <name>hbase.hregion.majorcompaction</name>
    <value>604800000</value>
</property>

##使用major_compact命令
major_compact tableName

当region⼤小⼤于某个阈值(hbase.hregion.max.filesize=10G)之后就会触发切分，⼀个region等分为2个region。
但是在生产线上这种切分策略却有相当大的弊端:
    切分策略对于⼤表和⼩表没有明显的区分。阈值(hbase.hregion.max.filesize)设置较大对大表⽐较友好，
  但是⼩表就有可能不会触发分裂，极端情况下可能就1个，这对业务来说并不是什么好事。
  如果设置较⼩则对小表友好，但一个⼤表就会在整个集群产⽣大量的region，这对于集群的管理、资源使用、failover来说都不是⼀件好事。

切分策略稍微有点复杂，总体看和ConstantSizeRegionSplitPolicy思路路相同，一个region⼤小⼤于设置阈值就会触发切分。
但是这个阈值并不像ConstantSizeRegionSplitPolicy是一个固定的值，⽽是会在一定条件下不断调整，调整则和region所属表在当前regionserver上的region个数有关系.
region split的计算公式是:
regioncount^3 * 128M * 2，当region达到该size的时候进行split
例如:
第⼀次split:1^3 * 256 = 256MB
第⼆次split:2^3 * 256 = 2048MB
第三次split:3^3 * 256 = 6912MB
第四次split:4^3 * 256 = 16384MB > 10GB，因此取较小的值10GB
后⾯每次split的size都是10GB了

这种切分策略的切分阈值又发⽣了变化，相⽐IncreasingToUpperBoundRegionSplitPolicy简单了一些，依然和待分裂region所属表在当前regionserver上的region个数有关系，
如果region个数等于1，切分阈值为flush size * 2，否则为MaxRegionFileSize。
这种切分策略对于⼤集群中的⼤表、⼩表会⽐IncreasingToUpperBoundRegionSplitPolicy更加友好，⼩表不会再产⽣大量的小region，而是适可⽽止。

根据rowKey的前缀对数据进⾏分组，这里是指定rowKey的前多少位作为前缀，⽐如rowKey都是16位的，指定前5位是前缀，那么前5位相同的rowKey在进行
region split的时候会分到相同的region中。

保证相同前缀的数据在同⼀个region中，例如rowKey的格式为:userid_eventtype_eventid，指定的delimiter为 _ ，则split的的时候会确保userid相同的数据在同一个region中。

<property>
    <name>hbase.regionserver.region.split.policy</name>
    <value>org.apache.hadoop.hbase.regionserver.IncreasingToUpperBoundRegionSplitPolicy</value>
</property>

HTableDescriptor tableDesc = new HTableDescriptor("test1");
tableDesc.setValue(HTableDescriptor.SPLIT_POLICY, IncreasingToUpperBoundRegionSplitPolicy.class.getName());
tableDesc.addFamily(new HColumnDescriptor(Bytes.toBytes("cf1")));
admin.createTable(tableDesc);

hbase> create 'test2', {METADATA => {'SPLIT_POLICY' => 'org.apache.hadoop.hbase.regionserver.IncreasingToUpperBoundRegionSplitPolicy'}},{NAME => 'cf1'}

create 'person','info1','info2',SPLITS => ['1000','2000','3000']

vim split.txt

aaa
bbb
ccc
ddd

create 'student','info',SPLITS_FILE => '/root/hbase/split.txt'

hbase org.apache.hadoop.hbase.util.Merge student \
student,,1595256696737.fc3eff4765709e66a8524d3c3ab42d59. \
student,aaa,1595256696737.1d53d6c1ce0c1bed269b16b6514131d0.

与冷合并不同的是，online_merge的传参是Region的hash值，⽽Region的hash值就是Region名称的最后那段在两个.之间的字符串部分。
需求:需要把lagou_s表中的2个region数据进行合并: 
student,,1587392159085.9ca8689901008946793b8d5fa5898e06. \ 
student,aaa,1587392159085.601d5741608cedb677634f8f7257e000.
 
需要进入hbase shell:
merge_region 'c8bc666507d9e45523aebaffa88ffdd6','02a9dfdf6ff42ae9f0524a3d8f4c7777'

 <dependencies>
    <dependency>
        <groupId>org.apache.hbase</groupId>
        <artifactId>hbase-client</artifactId>
        <version>1.3.1</version>
        </dependency>
    <dependency>
        <groupId>junit</groupId>
        <artifactId>junit</artifactId>
        <version>4.12</version>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.testng</groupId>
        <artifactId>testng</artifactId>
        <version>6.14.3</version>
        <scope>test</scope>
    </dependency>
</dependencies>

package com.lagou.hbase;
 
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
 
import java.io.IOException;
 
/**
 * @Author wanglitao
 * @Date 2020/8/4 10:47 下午
 * @Version 1.0
 * @Description
 */
public class HbaseClientDemo {
    Configuration conf = null;
    Connection conn = null;
    HBaseAdmin admin = null;
 
    @Before
    public void init() throws IOException {
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "linux121,linux122");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        conn = ConnectionFactory.createConnection(conf);
    }
 
    //创建一张hbase表
    @Test
    public void createTable() throws IOException {
        //获取HbaseAdmin对象用来创建表
        admin = (HBaseAdmin) conn.getAdmin();
        //创建Htabledesc描述器，表描述器
        HTableDescriptor worker = new HTableDescriptor(TableName.valueOf("worker"));
        //指定列族
        worker.addFamily(new HColumnDescriptor("info"));
        admin.createTable(worker);
        System.out.println("worker表创建成功！！");
    }
 
    //插入一条数据
    @Test
    public void putData() throws IOException {
        //需要获取一个table对象
        Table worker = conn.getTable(TableName.valueOf("worker"));
        //准备put对象
        Put put = new Put(Bytes.toBytes("110"));//指定rowkey
        put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("addr"), Bytes.toBytes("beijing"));
        //插入数据，参数类型是put
        worker.put(put);
        //准备list<puts>,可以执行批量插入
        //关闭table对象
        worker.close();
        System.out.println("插入数据到worker表成功！！");
    }
 
 
    //删除一条数据
    @Test
    public void deleteData() throws IOException {
        //需要获取一个table对象
        Table worker = conn.getTable(TableName.valueOf("worker"));
        //准备delete对象
        Delete delete = new Delete(Bytes.toBytes("110"));
        //执行删除
        worker.delete(delete);
        //关闭table对象
        worker.close();
        System.out.println("删除数据成功！！");
    }
 
    //查询数据
    @Test
    public void getData() throws IOException {
        //准备table对象
        Table worker = conn.getTable(TableName.valueOf("worker"));
        //准备get对象
        Get get = new Get(Bytes.toBytes("110"));
        //指定查询某个列族或者列
        get.addFamily(Bytes.toBytes("info"));
        //执行查询
        Result result = worker.get(get);
        //获取到result中所有cell对象
        Cell[] cells = result.rawCells();
        //遍历打印
        for (Cell cell : cells) {
            String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
            String f = Bytes.toString(CellUtil.cloneFamily(cell));
            String column = Bytes.toString(CellUtil.cloneQualifier(cell));
            String value = Bytes.toString(CellUtil.cloneValue(cell));
 
            System.out.println("rowkey-->" + rowkey + ";cf-->" + f + ";column-->" + column + ";value-->" + value);
        }
        worker.close();
    }
 
    //全表扫描
    @Test
    public void scanData() throws IOException {
        //准备table对象
        Table worker = conn.getTable(TableName.valueOf("worker"));
        //准备scan对象
        Scan scan = new Scan();
 
        //执行扫描
        ResultScanner resultScanner = worker.getScanner(scan);
        for (Result result : resultScanner) {
            //获取到result中所有cell对象
            Cell[] cells = result.rawCells();
            //遍历打印
            for (Cell cell : cells) {
                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                String f = Bytes.toString(CellUtil.cloneFamily(cell));
                String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println("rowkey-->" + rowkey + ";cf-->" + f + ";column-->" + column + ";value-->" + value);
            }
        }
        worker.close();
    }
 
    //指定scan 开始rowkey和结束rowkey，这种查询方式建议使用，指定开始和结束rowkey区间避免全表扫描
    @Test
    public void scanStartEndData() throws IOException {
        //准备table对象
        Table worker = conn.getTable(TableName.valueOf("worker"));
        //准备scan对象
        Scan scan = new Scan();
        //指定查询的rowkey区间,rowkey在hbase中是以字典序排序
        scan.setStartRow(Bytes.toBytes("001"));
        scan.setStopRow(Bytes.toBytes("004"));
        //执行扫描
        ResultScanner resultScanner = worker.getScanner(scan);
        for (Result result : resultScanner) {
            //获取到result中所有cell对象
            Cell[] cells = result.rawCells();
            //遍历打印
            for (Cell cell : cells) {
                String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                String f = Bytes.toString(CellUtil.cloneFamily(cell));
                String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println("rowkey=" + rowkey + ";cf=" + f + ";column=" + column + ";value=" + value);
            }
        }
        worker.close();
    }
 
    @After
    public void destroy() {
        if (admin != null) {
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

create 't1','info'
create 't2','info'

package com.lagou.coprocessor;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment; import org.apache.hadoop.hbase.filter.ByteArrayComparable;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import java.io.IOException;
public class MyProcessor extends BaseRegionObserver {
    @Override
    public void prePut(ObserverContext<RegionCoprocessorEnvironment> ce, Put put, WALEdit edit, Durability durability) throws IOException {
        HTableWrapper t2 = (HTableWrapper)ce.getEnvironment().getTable(TableName.valueOf("t2"));
        Cell nameCell = put.get("info".getBytes(), "name".getBytes()).get(0);
        Put put1 = new Put(put.getRow());
        put1.add(nameCell);
        t2.put(put);
        t2.close();
    }
}

<!-- https://mvnrepository.com/artifact/org.apache.hbase/hbase-server -->
<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-server</artifactId>
    <version>1.3.1</version>
</dependency>

cd /opt/lagou/softwares
mv original-hbaseStudy-1.0-SNAPSHOT.jar processor.jar
hdfs dfs -mkdir -p /processor
hdfs dfs -put processor.jar /processor

hbase(main):056:0> describe 't1'
# 1001表示优先级，该数字越小，优先级越高
hbase(main):055:0> alter 't1',METHOD => 'table_att','Coprocessor'=>'hdfs://linux121:9000/processor/processor.jar|com.lagou.hbase.processor.MyProcessor|1001|'
 
#再次查看't1'表，
hbase(main):043:0> describe 't1'

 put 't1','rk1','info:name','lisi'

disable 't1'
alter 't1',METHOD=>'table_att_unset',NAME=>'coprocessor$1'
enable 't2'

012,0,123,234,3.
0,3,012,123,234
0,012,123,234,3

4个region:[,a),[a,b),[b,c),[c,]
原始数据:abc1,abc2,abc3.
加盐后的rowkey:a-abc1,b-abc2,c-abc3
abc1,a
abc2,b

原始数据: abc1，abc2,abc3
哈希:
md5(abc1)=92231b....., 9223-abc1
md5(abc2) =32a131122...., 32a1-abc2
md5(abc3) = 452b1...., 452b-abc3.

<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>27.0.1-jre</version>
</dependency>

public static void main(String[] args) {
    // 1.创建符合条件的布隆过滤器
    // 预期数据量10000，错误率0.0001
    BloomFilter<CharSequence> bloomFilter = BloomFilter.create(Funnels.stringFunnel( Charset.forName("utf-8")),10000, 0.0001);
    // 2.将一部分数据添加进去
    for (int i = 0; i < 5000; i++) {
        bloomFilter.put("" + i); 
    }
    System.out.println("数据写入完毕");
    // 3.测试结果
    for (int i = 0; i < 10000; i++) {
        if (bloomFilter.mightContain("" + i)) {
            System.out.println(i + "存在");
        } else {
            System.out.println(i + "不存在");
        } 
    }
}