数据存储原理

hbase存储架构

img

  • 一个HRegionServer会负责管理很多个region
  • 一个==region==包含很多个==store==
    • 一个==列族==就划分成一个==store==
    • 如果一个表中只有1个列族,那么每一个region中只有一个store
    • 如果一个表中有N个列族,那么每一个region中有N个store
  • ==一个store==里面只有==一个memstore==
    • memstore是一块内存区域,写入的数据会先写入memstore进行缓冲,然后再把数据刷到磁盘
  • 一个store里面有很多个==StoreFile==, 最后数据是以很多个==HFile==这种数据结构的文件保存在HDFS上
    • StoreFile是HFile的抽象对象,如果说到StoreFile就等于HFile
    • ==每次memstore刷写数据到磁盘,就生成对应的一个新的HFile文件出来==

region

读数据流程

img

说明:HBase集群,只有一张meta表,此表只有一个region,该region数据保存在一个HRegionServer上

  • 1、客户端首先与zk进行连接;从zk找到meta表的region位置,即meta表的数据存储在某一HRegionServer上;客户端与此HRegionServer建立连接,然后读取meta表中的数据;meta表中存储了所有用户表的region信息,我们可以通过scan 'hbase:meta'来查看meta表信息
  • 2、根据要查询的namespace、表名和rowkey信息。找到写入数据对应的region信息
  • 3、找到这个region对应的regionServer,然后发送请求
  • 4、查找并定位到对应的region
  • 5、先从memstore查找数据,如果没有,再从BlockCache上读取
    • HBase上Regionserver的内存分为两个部分
      • 一部分作为Memstore,主要用来写;
      • 另外一部分作为BlockCache,主要用于读数据;
  • 6、如果BlockCache中也没有找到,再到StoreFile上进行读取
    • 从storeFile中读取到数据之后,不是直接把结果数据返回给客户端,而是把数据先写入到BlockCache中,目的是为了加快后续的查询;然后在返回结果给客户端。

写数据流程

img

  • 1、客户端首先从zk找到meta表的region位置,然后读取meta表中的数据,meta表中存储了用户表的region信息
  • 2、根据namespace、表名和rowkey信息。找到写入数据对应的region信息
  • 3、找到这个region对应的regionServer,然后发送请求
  • 4、把数据分别写到HLog(write ahead log)和memstore各一份
  • 5、memstore达到阈值后把数据刷到磁盘,生成storeFile文件
  • 6、删除HLog中的历史数据
补充:
HLog(write ahead log):
  也称为WAL意为Write ahead log,类似mysql中的binlog,用来做灾难恢复时用,HLog记录数据的所有变更,一旦数据修改,就可以从log中进行恢复。

flush、compact机制

Flush触发条件

  • memstore级别限制
<!--
    当Region中任意一个MemStore的大小达到了上限(hbase.hregion.memstore.flush.size,默认128MB),会触发Memstore刷新。
-->
<property>
    <name>hbase.hregion.memstore.flush.size</name>
    <value>134217728</value>
</property>
  • region级别限制
<!--
    当Region中所有Memstore的大小总和达到了上限(hbase.hregion.memstore.block.multiplier * hbase.hregion.memstore.flush.size,默认 2* 128M = 256M),会触发memstore刷新。
-->
<property>
    <name>hbase.hregion.memstore.flush.size</name>
    <value>134217728</value>
</property>
<property>
    <name>hbase.hregion.memstore.block.multiplier</name>
    <value>2</value>
</property>   
  • Region Server级别限制
<!--
  - 当一个Region Server中所有Memstore的大小总和超过低水位阈值hbase.regionserver.global.memstore.size.lower.limit*hbase.regionserver.global.memstore.size(前者默认值0.95),RegionServer开始强制flush;
  - 先Flush Memstore最大的Region,再执行次大的,依次执行;
  - 如写入速度大于flush写出的速度,导致总MemStore大小超过高水位阈值hbase.regionserver.global.memstore.size(默认为JVM内存的40%),此时RegionServer会阻塞更新并强制执行flush,直到总MemStore大小低于低水位阈值
-->
<property>
    <name>hbase.regionserver.global.memstore.size.lower.limit</name>
    <value>0.95</value>
</property>
<property>
    <name>hbase.regionserver.global.memstore.size</name>
    <value>0.4</value>
</property>
  • HLog数量上限

    • 当一个Region Server中HLog数量达到上限(可通过参数hbase.regionserver.maxlogs配置)时,系统会选取最早的一个 HLog对应的一个或多个Region进行flush
  • 定期刷新Memstore

    • 默认周期为1小时,确保Memstore不会长时间没有持久化。为避免所有的MemStore在同一时间都进行flush导致的问题,定期的flush操作有20000左右的随机延时。
  • 手动flush

    • 用户可以通过shell命令flush ‘tablename’或者flush ‘region name’分别对一个表或者一个Region进行flush。

Flush的流程

  • 为了减少flush过程对读写的影响,将整个flush过程分为三个阶段:

    • prepare阶段:遍历当前Region中所有的Memstore,将Memstore中当前数据集CellSkipListSet做一个快照snapshot;然后再新建一个CellSkipListSet。后期写入的数据都会写入新的CellSkipListSet中。prepare阶段需要加一把updateLock对写请求阻塞,结束之后会释放该锁。因为此阶段没有任何费时操作,因此持锁时间很短。

    • flush阶段:遍历所有Memstore,将prepare阶段生成的snapshot持久化为临时文件,临时文件会统一放到目录.tmp下。这个过程因为涉及到磁盘IO操作,因此相对比较耗时。

    • commit阶段:遍历所有Memstore,将flush阶段生成的临时文件移到指定的ColumnFamily目录下,针对HFile生成对应的storefile和Reader,把storefile添加到HStore的storefiles列表中,最后再清空prepare阶段生成的snapshot。

Compact合并机制

  • hbase为了==防止小文件过多==,以保证查询效率,hbase需要在必要的时候将这些小的store file合并成相对较大的store file,这个过程就称之为compaction。

minor compaction

  • 在将Store中多个HFile合并为一个HFile

    在这个过程中会选取一些小的、相邻的StoreFile将他们合并成一个更大的StoreFile,对于超过了TTL的数据、更新的数据、删除的数据仅仅只是做了标记。并没有进行物理删除,一次Minor Compaction的结果是更少并且更大的StoreFile。这种合并的触发频率很高。

  • minor compaction触发条件由以下几个参数共同决定:

<!--表示至少需要三个满足条件的store file时,minor compaction才会启动-->
<property>
    <name>hbase.hstore.compactionThreshold</name>
    <value>3</value>
</property>

<!--表示一次minor compaction中最多选取10个store file-->
<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>

major compaction

  • 合并Store中所有的HFile为一个HFile

    将所有的StoreFile合并成一个StoreFile,这个过程还会清理三类无意义数据:被删除的数据、TTL过期数据、版本号超过设定版本号的数据。合并频率比较低,默认7天执行一次,并且性能消耗非常大,建议生产关闭(设置为0),在应用空闲时间手动触发。一般可以是手动控制进行合并,防止出现在业务高峰期。

  • major compaction触发时间条件

    <!--默认值为7天进行一次大合并,-->
    <property>
        <name>hbase.hregion.majorcompaction</name>
        <value>604800000</value>
    </property>
    
  • 手动触发

    ##使用major_compact命令
    major_compact tableName
    

region 拆分机制

  • region中存储的是大量的rowkey数据 ,当region中的数据条数过多的时候,直接影响查询效率.当region过大的时候.hbase会拆分region , 这也是Hbase的一个优点 .

  • HBase的region split策略一共有以下几种:

  • 1、ConstantSizeRegionSplitPolicy

    • 0.94版本前默认切分策略
  • 当region大小大于某个阈值(hbase.hregion.max.filesize=10G)之后就会触发切分,一个region等分为2个region。

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

    • 0.94版本~2.0版本默认切分策略
    • 切分策略稍微有点复杂,总体看和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了

  • 3、SteppingSplitPolicy

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

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

    • 保证相同前缀的数据在同一个region中,例如rowKey的格式为:userid_eventtype_eventid,指定的delimiter为 _ ,则split的的时候会确保userid相同的数据在同一个region中。
  • 6、DisabledRegionSplitPolicy
    • 不启用自动拆分, 需要指定手动拆分

表的预分区

  • 当一个table刚被创建的时候,Hbase默认的分配一个region给table。也就是说这个时候,所有的读写请求都会访问到同一个regionServer的同一个region中,这个时候就达不到负载均衡的效果了,集群中的其他regionServer就可能会处于比较空闲的状态。
    • 解决这个问题可以用pre-splitting,在创建table的时候就配置好,生成多个region。
  • 为何要预分区?

    • 增加数据读写效率
    • 负载均衡,防止数据倾斜
    • 方便集群容灾调度region
    • 优化Map数量
  • 每一个region维护着startRow与endRowKey,如果加入的数据符合某个region维护的rowKey范围,则该数据交给这个region维护。

-- 创建表,指定预分区
create 'person','info1','info2',SPLITS => ['1000','2000','3000','4000']

-- 根据文件内容预分区
cat /kfly/install/split.txt
  aaa
  bbb
  ccc
  ddd
create 'student','info',SPLITS_FILE => '/kfly/install/split.txt'
  • HexStringSplit 算法

    • HexStringSplit会将数据从“00000000”到“FFFFFFFF”之间的数据长度按照n等分之后算出每一段的其实rowkey和结束rowkey,以此作为拆分点。
    create 'mytable', 'base_info',' extra_info', {NUMREGIONS => 15, SPLITALGO => 'HexStringSplit'}
    

hbasePreSplit

region 合并

  • Region的合并不是为了性能, 而是出于维护的目的 .
  • 比如删除了大量的数据 ,这个时候每个Region都变得很小 ,存储多个Region就浪费了 ,这个时候可以把Region合并起来,进而可以减少一些Region服务器节点
  • 通过Merge类冷合并Region

    • 执行合并前,==需要先关闭hbase集群==

    • 创建一张hbase表:

create 'test','info1',SPLITS => ['1000','2000','3000']
  • 查看表region

testRegion

  • 需求:
  • 这里通过org.apache.hadoop.hbase.util.Merge类来实现,不需要进入hbase shell,直接执行(==需要先关闭hbase集群==):
  # 需要把test表中的2个region数据进行合并:test,,1565940912661.62d28d7d20f18debd2e7dac093bc09d8.  test,1000,1565940912661.5b6f9e8dad3880bcc825826d12e81436.

  hbase org.apache.hadoop.hbase.util.Merge test test,,1565940912661.62d28d7d20f18debd2e7dac093bc09d8. test,1000,1565940912661.5b6f9e8dad3880bcc825826d12e81436.
  • 成功后界面观察

testMerge

  • 通过online_merge热合并Region

    • ==不需要关闭hbase集群==,在线进行合并

    • 与冷合并不同的是,online_merge的传参是Region的hash值,而Region的hash值就是Region名称的最后那段在两个.之间的字符串部分。

    • 需求:需要把test表中的2个region数据进行合并:
      test,2000,1565940912661.c2212a3956b814a6f0d57a90983a8515.
      test,3000,1565940912661.553dd4db667814cf2f050561167ca030.

    • 需要进入hbase shell:

    merge_region 'c2212a3956b814a6f0d57a90983a8515','553dd4db667814cf2f050561167ca030'
    
    • 成功后观察界面

online_merge

HBase集成MapReduce

  • HBase表中的数据最终都是存储在HDFS上,HBase天生的支持MR的操作,我们可以通过MR直接处理HBase表中的数据,并且MR可以将处理后的结果直接存储到HBase表中。

  • 需求:==读取HBase当中myuser这张表的数据,将数据写入到另外一张myuser2表里面去==

 <dependency>
   <groupId>org.apache.hadoop</groupId>
   <artifactId>hadoop-client</artifactId>
   <version>2.6.0-mr1-cdh5.14.2</version>
</dependency>
<dependency>
  <groupId>org.apache.hbase</groupId>
  <artifactId>hbase-client</artifactId>
  <version>1.2.0-cdh5.14.2</version>
</dependency>
<dependency>
  <groupId>org.apache.hbase</groupId>
  <artifactId>hbase-server</artifactId>
  <version>1.2.0-cdh5.14.2</version>
</dependency>
class HBaseMapper extends TableMapper<Text,Put>{
  protected void map(ImmutableBytesWritable key, Result value, Context context){
    // key get rowKey
    // result -> put
    context.write(new Text(rowkey),put);
  }
}
class HbaseReducer extends TableReducer<Text,Put,ImmutableBytesWritable>{
  protected void reduce(Text key, Iterable<Put> values, Context context){
    for (Put put : values) {
      context.write(null,put);
    }
  }
}
public static void main(String[] args){
        Configuration conf = new Configuration();
        Scan scan = new Scan();
        Job job = Job.getInstance(conf);
        job.setJarByClass(HBaseMR.class);
        //使用TableMapReduceUtil 工具类来初始化我们的mapper
      TableMapReduceUtil.initTableMapperJob(
                                            TableName.valueOf(args[0]),
                                            scan,
                                            HBaseMapper.class,
                                            Text.class,
                                            Put.class,
                                            job
                                                                              );
        //使用TableMapReduceUtil 工具类来初始化我们的reducer
        TableMapReduceUtil.initTableReducerJob(
                                                                      args[1],
                                                                      HbaseReducer.class,
                                                                      job);
        //设置reduce task个数
        job.setNumReduceTasks(1);
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
# 打成jar包提交到集群中运行
hadoop jar hbase_java_api-1.0-SNAPSHOT.jar com.kaikeba.HBaseMR t1 t2
  • 需求

    • ==通过bulkload的方式批量加载数据到HBase表中==
    • ==将我们hdfs上面的这个路径/hbase/input/user.txt的数据文件,转换成HFile格式,然后load到myuser2这张表里面去==
  • 知识点描述

    • 加载数据到HBase当中去的方式多种多样,我们可以使用HBase的javaAPI或者使用sqoop将我们的数据写入或者导入到HBase当中去,但是这些方式不是慢就是在导入的过程的占用Region资源导致效率低下
    • 我们也可以通过MR的程序,将我们的数据直接转换成HBase的最终存储格式HFile,然后直接load数据到HBase当中去即可
  • HBase数据正常写流程回顾

    hbase-write

  • bulkload方式的处理示意图

  • 好处

    • 导入过程不占用Region资源
    • 能快速导入海量的数据
    • 节省内存
  • ==1、开发生成HFile文件的代码==

// 1. map阶段
context.write(new ImmutableBytesWritable(Bytes.toBytes(split[0])),put);

// 2. main job
Configuration conf = HBaseConfiguration.create();
Connection connection = ConnectionFactory.createConnection(conf);
job.setMapOutputKeyClass(ImmutableBytesWritable.class);
job.setMapOutputValueClass(Put.class);
//指定输出的类型HFileOutputFormat2
job.setOutputFormatClass(HFileOutputFormat2.class);
HFileOutputFormat2
  .configureIncrementalLoad(
                            job,
                            table,                                                       
                            conn.getRegionLocator(TableName.valueOf("t4"))
                                                            );
// 3. 加载数据使用java Api
Table table = connection.getTable(tableName);
//构建LoadIncrementalHFiles加载HFile文件
LoadIncrementalHFiles load = new LoadIncrementalHFiles(configuration);
load.doBulkLoad(new Path("hdfs://node01:8020/hbase/output_file");

// 加载数据命令加载,先将hbase的jar包添加到hadoop的classpath路径下
export HBASE_HOME=/kfly/install/hbase-1.2.0-cdh5.14.2/
export HADOOP_HOME=/kfly/install/hadoop-2.6.0/
export HADOOP_CLASSPATH=`${HBASE_HOME}/bin/hbase mapredcp`

yarn jar /kfly/install/hbase-1.2.0-cdh5.14.2/lib/hbase-server-1.2.0-cdh5.14.2.jar   completebulkload /hbase/output_hfile myuser2
hadoop jar hbase_java_api-1.0-SNAPSHOT.jar com.kaikeba.HBaseLoad

HBase集成Hive

  • Hive提供了与HBase的集成,使得能够在HBase表上使用hive sql 语句进行查询、插入操作以及进行Join和Union等复杂查询,同时也可以将hive表中的数据映射到Hbase中

对比

  • Hive

    • 数据仓库

      • ​ Hive的本质其实就相当于将HDFS中已经存储的文件在Mysql中做了一个双射关系,以方便使用HQL去管理查询。
    • 用于数据分析、清洗

      • ​ Hive适用于离线的数据分析和清洗,延迟较高
    • 基于HDFS、MapReduce

      • ​ Hive存储的数据依旧在DataNode上,编写的HQL语句终将是转换为MapReduce代码执行。(不要钻不需要执行MapReduce代码的情况的牛角尖)
  • HBase

    • 数据库

      • 是一种面向列存储的非关系型数据库。
    • 用于存储结构化和非结构话的数据

      • 适用于单表非关系型数据的存储,不适合做关联查询,类似JOIN等操作。
    • 基于HDFS

      • 数据持久化存储的体现形式是Hfile,存放于DataNode中,被ResionServer以region的形式进行管理。
    • 延迟较低,接入在线业务使用

      • 面对大量的企业数据,HBase可以直线单表大量数据的存储,同时提供了高效的数据访问速度。
  • Hive和Hbase是两种基于Hadoop的不同技术,Hive是一种类SQL的引擎,并且运行MapReduce任务,Hbase是一种在Hadoop之上的NoSQL 的Key/vale数据库。这两种工具是可以同时使用的。就像用Google来搜索,用FaceBook进行社交一样,Hive可以用来进行统计查询,HBase可以用来进行实时查询,数据也可以从Hive写到HBase,或者从HBase写回Hive。

拷贝jar包

  • 将我们HBase的五个jar包拷贝到hive的lib目录下

  • hbase的jar包都在/kfly/install/hbase-1.2.0-cdh5.14.2/lib

  • 我们需要拷贝五个jar包名字如下

hbase-client-1.2.0-cdh5.14.2.jar                  
hbase-hadoop2-compat-1.2.0-cdh5.14.2.jar 
hbase-hadoop-compat-1.2.0-cdh5.14.2.jar  
hbase-it-1.2.0-cdh5.14.2.jar    
hbase-server-1.2.0-cdh5.14.2.jar
  • 我们直接在node03执行以下命令,通过创建软连接的方式来进行jar包的依赖
ln -s /kfly/install/hbase-1.2.0-cdh5.14.2/lib/hbase-client-1.2.0-cdh5.14.2.jar              /kfly/install/hive-1.1.0-cdh5.14.2/lib/hbase-client-1.2.0-cdh5.14.2.jar   

修改hive-site.xml

  • 添加以下两个属性的配置
<property>
        <name>hive.zookeeper.quorum</name>
        <value>node01,node02,node03</value>
</property>
 <property>
        <name>hbase.zookeeper.quorum</name>
        <value>node01,node02,node03</value>
</property>

修改hive-env.sh

export HADOOP_HOME=/export/servers/hadoop-2.6.0
export HBASE_HOME=/export/servers/hbase-1.2.0-cdh5.14.2
export HIVE_CONF_DIR=/export/servers/hive-1.1.0-cdh5.14.2/conf

hive建表同步到hbase

-- hive当中建表
create external table if not exists course.score(id int,cname string,score int) row format delimited fields terminated by '\t' stored as textfile ;
-- 加载数据到hive
load data local inpath '/kfly/doc/hive-hbase.txt' into table score;
-- 创建hive管理表与HBase映射
create table course.hbase_score(id int,cname string,score int) stored by 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'  with serdeproperties("hbase.columns.mapping" = "cf:name,cf:score") tblproperties("hbase.table.name" = "hbase_score");

hive外部表映射HBase表模型

-- 创建一张hbase表
create 'hbase_hive_score',{ NAME =>'cf'}
-- 建立hive的外部表,映射HBase当中的表以及字段
CREATE external TABLE course.hbase2hive(id int, name string, score int) STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler' WITH SERDEPROPERTIES ("hbase.columns.mapping" = ":key,cf:name,cf:score") TBLPROPERTIES("hbase.table.name" ="hbase_hive_score");