2021-12-22 迈向程序猿的第五十一步

目录

一.Hbase的工作机制

1.1 Hbase的寻址机制(重点)

1.2 Hbase的存储机制(重中之重)

1.2.1 存储机制介绍

1.2.2 名词解释

1.3 Hbase的Region管理

1.4 Hbase的读写流程(重点)

1.4.1 Hbase的读流程

1.4.2 Hbase的写流程

1.5 布隆过滤器

1.5.1 简介

1.5.2 原理

1.5.3 Hbase中的布隆过滤器的设置 

二.Hbase与Hive,Mapreduce的整合

2.1 Hbase与Hive的整合

2.1.1 为什么要整合

2.1.2 Hive-to-hbase

2.1.3 Hbase-to-Hive

2.1.4 总结

三.Hbase的二级索引和协处理器

3.1 二级索引的简介

3.2 协处理器组件简介

3.2.1 为什么要引入协处理器

3.2.2 协处理器的分类(熟悉)

3.2.3 协处理器的加载和卸载

3.3 案例演示:使用协处理器完成二级索引表的创建

3.3.1 案例描述

3.3.2 代码编写

3.3.3 动态加载协处理器

3.3.4 案例测试


一.Hbase的工作机制

1.1 Hbase的寻址机制(重点)

1.   table.put()      put 'myns:student'
     table.get()/getScanner()      get/scan....
     
     put 'myns:student', 'rk01000', 'f1:name'
       
       
     region:  startkey,   endkey
     
-- zookeeper:  维护了meta表的信息,就是地址
               get /hbase/meta-region-server
-- hbase:meta表:    记录了所有表的所有的region信息
                每个region以四个单元格进行记录,单元格的rowkey,就是region的名称
                四个单元格中的两个单元格:
                info:regioninfo维护的是region的名称,以及行范围
                info:server维护的region的位置
                
                region名称: schema:tablename,startkey,timestamp.ENCODED

1.2 Hbase的存储机制(重中之重)

1.2.1 存储机制介绍

1.  通过寻址流程定位到具体的region。
2. 将单元格存储到store对应的memstore中
3. 排序:按照rowkey进行升序,key进行升序,timestamp降序
        单元格数据格式如下:  
        rowkey:column family:column:value:timestamp
        
4. 达到flush的阈值时,开始flush,生成storefile.然后由store来维护n个storefile的索引信息,比如path,startkey,endkey
      
      flush阈值: 128M、 1小时、  当前region内存的40%  、当前regionserver内存40%
      
5. 当storefile的数量达到阈值,比如是3个。就进行合并。  合并期间会进行排序(rowkey升序,column升序,timetamp降序),  还会进行真正的删除(当发现有delete标记的单元格,就将所有版本过滤掉)和修改操作(超过版本数量的老版本过滤掉)。
​
6. 当合并的文件越来越大时,如果达到阈值(10G),就会触发split操作,数据尽量做到均分(会影响其他文件的切分),从而造成了region的切分。  旧region下线,两个的新的region维护数据,由hmaster来重新进行负载均衡。        

1.2.2 名词解释

flush

当memstore达到阈值是,将内存中的数据冲刷成文件
​
可以手动flush
查看用法:  help 'flush'

compact

当storefile达到数量阈值时,进行合并操作。
​
手动合并:
major_compact  regionname

split

当文件大小达到阈值时,会造成region的切分
手动切分:
split regionname,splitkey

1.3 Hbase的Region管理

1)预切分

建表期间,进行预切分,指的就是提前划分region。
create tablename,column family,SPLITS=>[startkey1,startkey2....] 
region数量 = startkey的数量+1

2)当文件达到阈值时,会主动切分

split regionname,splitkey

3)region也可以合并

merge_region  'encoded_region','encoded_region'

1.4 Hbase的读写流程(重点)

1.4.1 Hbase的读流程

1.  客户端请求zookeeper,获取meta表的位置信息
2.  客户端跳转到meta表,获取要访问的表的具体region位置
3.  客户端跳转到具体region所在的regionserver,找到该region。
4.  访问对应的store下的memstore(写缓存), 如果有数据就返回,如果没有数据,就访问regionserver对应的读缓存,如果没有数据,再访问磁盘,然后数据返回给客户端,并保存到读缓存中,方便下次快速读取。

1.4.2 Hbase的写流程

1.  客户端请求zookeeper,获取meta表的位置信息
2.  客户端跳转到meta表,获取要访问的表的具体region位置
3.  客户端跳转到具体region所在的regionserver,找到该region,将单元格写入到对应的store里的memstore
4.  在memstore里进行排序(按照rowkey进行升序,key进行升序,timestamp降序)
5.  如果达到memstore的阈值,就会flush成storefile文件,由store来维护该文件的索引信息。
6.  如果storefile数量达到阈值是,会进行合并操作(排序,实际的删除和修改)
7.  如果合并成的文件达到大小阈值时,会进行切分,造成region的切分(旧region下线,两个新region生成)

1.5 布隆过滤器

1.5.1 简介

- 布隆过滤器是一个数据结构
- 内部维护着一个二进制向量的位数组,默认大小64K
- 还维护着N个hash算法, 默认值是3个。
- 牺牲了准确率,算法的时间复杂度是O(1)。从而提高了查询效率
- 特点:  判断一个元素是否在一个集合中,有两种结果:一种就是在集合中,但是不一定真实存在
         另外一种结果就是不在集合中,那么就一定不在集合中。

1.5.2 原理

当存储元素时,会调用hash算法,计算元素的三个hash值,三个值作为bit数组的下标,将对应的元素由0改为1. (注意,hbase在put数据,不判断是否存在过)。
​
当在查询一个元素是否在这个集合中时,也是算出三个hash值作为下标,访问对应上的数据是否为1,如果全都是1,表明该元素可能存在这个集合中。只要有一个位置上是0,则表示该元素一定不在集合中。 

1.5.3 Hbase中的布隆过滤器的设置 

方法:columnfamily.setBloomFilterType()
布隆过滤器的级别:
  BloomType.NONE:    表示不启用布隆过滤器
  BloomType.ROW:     行级别的,  只对每一个rowkey做 布隆过滤器的数据存储
  BloomType.ROWCOL:  行列级别的, 对rowkey:columnfamily:column做布隆过滤器的数据存储
  
  
  参考文档上的布隆过滤器章节的作用:

二.Hbase与Hive,Mapreduce的整合

2.1 Hbase与Hive的整合

2.1.1 为什么要整合

hbase: hadoop数据库,本质用来存储大数据集,虽然提供了近似实时的读写功能
hive:  数据仓库的管理工具,作用就是用来使用hql语言对数据进行分析。

2.1.2 Hive-to-hbase

在hive创建表, 可以在hbase中看见
create table if not exists employee (
uid int,
uname string,
age int,
sex string,
province string
)
stored by 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'
with serdeproperties(
"hbase.columns.mapping"=":key,base_info:name,base_info:age,base_info:sex,address:provice"
)
tblproperties(
"hbase.table.name"="myns:employee"
);

动态加载数据

insert into employee values(1, 'michael', 32, '男','jilin');
insert into employee values(2,'wcm',23,'男','heilongjiang');
put 'myns:employee','3','base_info:name','lisi'

结果查看

- 在hive中写一个select语句
- 在50070webui上查看hive下的表目录,  结论:没有数据, hive不负责存储
- 在hbase中写一个scan语句
- 在50070webui上查看hbase下的表目录,  结论:应该有数据,如果看不到,是因为还在内存中

2.1.3 Hbase-to-Hive

hbase中先有表,然后映射到hive中
create 'myns:student','f1','f2'
​
put 'myns:student','rk00001','f1:name','zhaoyun'
put 'myns:student','rk00001','f1:age',23
put 'myns:student','rk00001','f1:gender','m'
put 'myns:student','rk00002','f1:name','zhenji'
put 'myns:student','rk00002','f1:age',24
put 'myns:student','rk00002','f2:province','hebei'

在hive中创建表与hbase中的表进行映射

drop table student;
create external table if not exists student (
sid string,
name string,
province string,
gender string,
age int
)
stored by 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'
with serdeproperties(
"hbase.columns.mapping"=":key,f1:name,f2:province,f1:gender,f1:age"
)
tblproperties(
"hbase.table.name"="myns:student"
);

结果查看

在hive中 写select语句进行查看。

2.1.4 总结

-1.hive表的字段与hbase表的单元格 是按照顺序映射,而不是根据名称映射。
-2.hbase的rowkey 可以映射成hive中的一个字段,通过:key进行映射。可以不写:key,则与hive的第一个字段映射
-3.在字段映射时,注意字段类型的合理性
-4.如果hbase表已经存在,进行hive新表映射,hive表必须是外部表。

三.Hbase的二级索引和协处理器

3.1 二级索引的简介

rk1001     f1:name-zhangsan      f1:age-23           f2:province-guangdong
........................
rk2001     f1:name-gaoyuanyuan      f1:age-38        f2:province-shanghai
...........
rk3001     f1:name-gaoyuanyuan
​
​
需求:  查询 name叫gaoyuanyuan的年龄
​
    SingleColumnValueFilter---    name:gaoyuanyuan
    
    hbase的底层逻辑: 遍历数据块中的所有行信息,要查看是否有name:gaoyuanyuan单元格,有,返回所有行。
           程序员需要再次编程对返回的所有行数据,查看是否age单元格。有,就返回。 性能很低。
           
           
如何提高上述需求类型的查询性能???  再维护一张表:单元格与rowkey的映射关系表。
​
index表:
rowkey:
.........
f1:name-gaoyuanyuan     info:rk1  'rk2001'
                        info:rk2  'rk3001'
......
f1:name-zhangsan        info:rk1  'rk1001'
........
​

什么是二级索引表

概念:维护的数据是另外一张表的单元格与rowkey的映射关系的表,就是二级索引表
作用:提高查询效率,避免对原表的全表遍历, 牺牲磁盘空间,换取查询效率。

小贴士:

二级索引表,应该是程序自动维护的。

3.2 协处理器组件简介

3.2.1 为什么要引入协处理器

在Hbase的低版本(0.92以前)作为列族数据库最经常被人诟病的特性包括:无法轻易建立“二级索引”,难以执行求和、计数、排序等操作。
之后引入了协处理器(coprocessor)进行改进,可以轻松的实现上述需求。

3.2.2 协处理器的分类(熟悉)

分两大类型:一个是Observer类型, 一个是endpoint类型
​
- Observer 允许集群在正常的客户端操作过程中可以有不同的行为表现  
- Observer 类似于 RDBMS 中的触发器,主要在服务端工作
- Observer 可以实现权限管理、优先级设置、监控、ddl 控制、二级索引等功能
​
- Endpoint 类似于 RDBMS 中的存储过程,主要在服务端工作
- Endpoint 允许扩展集群的能力,对客户端应用开放新的运算命令  
- Endpoint 可以实现 min、max、avg、sum、distinct、group by 等功能

3.2.3 协处理器的加载和卸载

1)加载方式:两种

第一种:静态加载,也叫系统级别的协处理器,只需要在hhbase-site.xml里添加如下属性和具体类名

<property>
   <name>hbase.coprocessor.user.region.classes</name>
   <value>类全名</value>
</property>
​
可以用”,”分割加载多个 class

第二种:动态记载,称之为表级别的协处理器

只对特定的表生效。通过 HBase Shell 来实现。
​
1. 停用表  disable 'mytable'
​
2. 添加协处理器  
alter 't_guanzhu',METHOD => 'table_att',
'coprocessor'=>
'hdfs://supercluster/jar/mycoprocessor.jar|com.xx.hbase.coprocessor.MyIndexCoprocessor|1001|'
3. 启用表  enable 'mytable'

2)卸载方式

1. disable 'mytable'
2. alter 'mytable',METHOD=>'table_att_unset',NAME=>'coprocessor$1'
3. enable 'mytable'

3)注意事项:

如果写的协处理器逻辑有问题,那么可能会造成hbase的集群宕机
​
解决办法:
1. 关闭所有的hbase的守护进程
2. 进入zookeeper,删除相关的znode。  如果不知道是哪一个znode,就将hbase 整个删掉
3. 重新启动hbase,删除掉挂载了协处理器的那张表。重新维护表
4. 再加载协处理器,进行测试

3.3 案例演示:使用协处理器完成二级索引表的创建

3.3.1 案例描述

模拟博客上的关注信息表,以及粉丝表。 
1. 关注信息表是原表: 维护的是每个用户关注的人
      rk0001     f1:user-wcm    f1:obj-gaoyuanyuan    ......
      ......
      rk0009     f1:user-wcm    f1:obj-canglaoshi     .....
      ....
      rk0019     f1:user-laoli  f1:liuyifei
    
2. 粉丝表是二级索引表。
   
        wcm-gaoyuanyuan      f1:rk-rk0001
        wcm-canglaoshi       f1:rk-rk0009
        ...........
        laoli-gaoyuanyuan    f1:rk-rk0019
        
create 'fans','f1'      

3.3.2 代码编写

package com.xx.hbase.coprocessor;
​
import com.xx.hbase.util.HbaseUtil;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
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.regionserver.wal.WALEdit;
​
import java.io.IOException;
import java.util.List;
​
public class FansObServer extends BaseRegionObserver {
    /**
     * 重写prePut方法。
     * @param e
     * @param put    put形参就会主动接受客户端提交的put对象
     * @param edit
     * @param durability
     * @throws IOException
     *
     *      rk0001     f1:user-wcm    f1:obj-gaoyuanyuan
     *
     *
     *      注意:该Observer是对原表进行监听
     */
    @Override
    public void prePut(ObserverContext<RegionCoprocessorEnvironment> e, Put put, WALEdit edit, Durability durability) throws IOException {
        //获取put对象上的rowkey
        byte[] row = put.getRow();
        String rowkey = new String(row);  //rk0001
        List<Cell> users = put.get("f1".getBytes(), "user".getBytes());
        Cell cell = users.get(0);
        String value = new String(CellUtil.cloneValue(cell));  //wcm
​
        List<Cell> objs = put.get("f1".getBytes(), "obj".getBytes());
        Cell cell2 = objs.get(0);
        String value1 = new String(CellUtil.cloneValue(cell2));  //gaoyuanyuan
​
        //先封装一个新的put对象,准备提交到Fans表中   wcm-gaoyuanyuan      f1:rk-rk0001
        Put newPut = new Put((value+"-"+value1).getBytes());
        newPut.addColumn("f1".getBytes(),"rk".getBytes(),rowkey.getBytes());
​
        Table table = HbaseUtil.getTable("Fans");  //千万别写错表名
        table.put(newPut);
        table.close();
​
    }
}

3.3.3 动态加载协处理器

--1. 创建关注表
create 'guanzhu','f1'
​
--2. 加载协处理器
     (1)先将jar包上传到hdfs上的/jar目录下
     (2)挂载
alter 'hello',METHOD => 'table_att','coprocessor'=>'hdfs://xxx01/jar/hello3_hbase.jar|com.xx.hbase.coprocessor.Hello2ObServer|1001|'

3.3.4 案例测试

小贴士:基于案例的需求,user和obj单元格应该同时添加,所以,应该使用api添加数据

@Before
public void getTable(){
   table = HbaseUtil.getTable("guanzhu");
}
@Test
public void putOne() throws IOException {
   //1. 获取Put对象,指定rowkey
   Put put = new Put(Bytes.toBytes("rk0001"));
   //2. 指定列族名,列名,列值
   put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("user"), Bytes.toBytes("wcm"));
   put.addColumn(Bytes.toBytes("f1"), Bytes.toBytes("obj"), Bytes.toBytes("gaoyuanyuan"));
   //3. 提交
   table.put(put);
}

最后查看两张表的信息

hbase(main):009:0> scan 'Fans'
ROW                              COLUMN+CELL
wcm-gaoyuanyuan                  column=f1:rk, timestamp=1640165499598, value=rk0001
​
​
​
hbase(main):010:0> scan 'guanzhu'
ROW                               COLUMN+CELL
 rk0001                           column=f1:obj, timestamp=1640165499610, value=gaoyuanyuan
 rk0001                           column=f1:user, timestamp=1640165499610, value=wcm

(=-=,这里写协助器的时候一定要注意逻辑,万一错了,hbase集群就直接瘫痪,处理起来很是麻烦!!!)

本图文内容来源于网友网络收集整理提供,作为学习参考使用,版权属于原作者。
THE END
分享
二维码
< <上一篇
下一篇>>