HBase学习笔记

HBase

HBase概念

HBase是一个分布式的、面向列的开源数据库，该技术来源于 Fay Chang 所撰写的Google论文“Bigtable：一个结构化数据的分布式存储系统”。就像Bigtable利用了Google文件系统（File System）所提供的分布式数据存储一样，HBase在Hadoop之上提供了类似于Bigtable的能力。HBase是Apache的Hadoop项目的子项目。HBase不同于一般的关系数据库，它是一个适合于非结构化数据存储的数据库。另一个不同的是HBase基于列的而不是基于行的模式。

HBase特点

• 强读写一致，但是不是“最终一致性”的数据存储，这使得它非常适合高速的计算聚合

• 自动分片，通过Region分散在集群中，当行数增长的时候，Region也会自动的切分和再分配

• 自动的故障转移

• Hadoop/HDFS集成，和HDFS开箱即用，不用太麻烦的衔接

• 丰富的“简洁，高效”API，Thrift/REST API，Java API

• 块缓存，布隆过滤器，可以高效的列查询优化

• 操作管理，Hbase提供了内置的web界面来操作，还可以监控JMX指标

HBase架构

Hbase架构如图1所示：

从图中可以看出Hbase是由Client、Zookeeper、Master、HRegionServer、HDFS等几个组件组成，下面来介绍一下几个组件的相关功能：

1. Client

   Client包含了访问Hbase的接口，另外Client还维护了对应的cache来加速Hbase的访问，比如cache的.META.元数据的信息。

2. Zookeeper

   HBase通过Zookeeper来做master的高可用、RegionServer的监控、元数据的入口以及集群配置的维护等工作。具体工作如下：

   通过Zoopkeeper来保证集群中只有1个master在运行，如果master异常，会通过竞争机制产生新的master提供服务。

   通过Zoopkeeper来监控RegionServer的状态，当RegionSevrer有异常的时候，通过回调的形式通知Master RegionServer上下线的信息。

   通过Zoopkeeper存储元数据的统一入口地址。

3. Hmaster

   master节点的主要职责如下：

   • 为RegionServer分配Region
   • 维护整个集群的负载均衡
   • 维护集群的元数据信息
   • 发现失效的Region，并将失效的Region分配到正常的RegionServer上
   • 当RegionSever失效的时候，协调对应Hlog的拆分

4. HRegionServer

   HRegionServer直接对接用户的读写请求，是真正的“干活”的节点。它的功能概括如下：

   • 管理master为其分配的Region
   • 处理来自客户端的读写请求
   • 负责和底层HDFS的交互，存储数据到HDFS
   • 负责Region变大以后的拆分
   • 负责Storefile的合并工作

5. HDFS

   HDFS为Hbase提供最终的底层数据存储服务，同时为HBase提供高可用（Hlog存储在HDFS）的支持，具体功能概括如下：

   • 提供元数据和表数据的底层分布式存储服务
   • 数据多副本，保证的高可靠和高可用性

HBase中的角色

HMaster

功能：

1. 监控RegionServer
2. 处理RegionServer故障转移
3. 处理元数据的变更
4. 处理region的分配或转移
5. 在空闲时间进行数据的负载均衡
6. 通过Zookeeper发布自己的位置给客户端

RegionServer

功能：

1. 负责存储HBase的实际数据
2. 处理分配给它的Region
3. 刷新缓存到HDFS
4. 维护HLog
5. 执行压缩
6. 负责处理Region分片

其他组件

1. Write-Ahead-logs

   HBase的修改记录，当对HBase读写数据的时候，数据不是直接写进磁盘，它会在内存中保留一段时间（时间以及数据量阈值可以设定）。但把数据保存在内存中可能有更高的概率引起数据丢失，为了解决这个问题，数据会先写在一个叫做Write-Ahead logfile的文件中，然后再写入内存中。所以在系统出现故障的时候，数据可以通过这个日志文件重建。

2. Region

   Hbase表的分片，HBase表会根据RowKey值被切分成不同的region存储在RegionServer中，在一个RegionServer中可以有多个不同的region。

3. Store

   HFile存储在Store中，一个Store对应HBase表中的一个列族。

4. MemStore

   顾名思义，就是内存存储，位于内存中，用来保存当前的数据操作，所以当数据保存在WAL中之后，RegsionServer会在内存中存储键值对。

5. HFile

   这是在磁盘上保存原始数据的实际的物理文件，是实际的存储文件。StoreFile是以HFile的形式存储在HDFS的。

HBase安装（高可用）

==需要jdk、hadoop、zookeeper的支持。==

下载安装Hbase

官网

安装命令：

tar -zxvf hbase-1.3.2-bin.tar.gz -C /usr/local/

将hbase解压安装至/usr/local目录下。

开启hadoop集群和zookeeper服务

启动命令：

[root@hadoop01 ~]# start-all.sh //hadoop集群启动
[root@hadoop01 ~]# zkServer.sh start //zookeeper服务启动 所有机器都得启动

使用JPS命令查看集群启动状态

hadoop01(namenode)

hadoop02(datanode)

hadoop03(datanode)

修改配置文件

配置hbase的全局变量

[root@hadoop01 ~]# vi /etc/profile
#加入hbase的环境变量
export HBASE_HOME=/usr/local/hbase-1.3.2
export PATH=$PATH:$HBASE_HOME/bin
[root@hadoop01 ~]# source /etc/profile

修改hbase-env.sh

[root@hadoop01 ~]#vi hbase-env.sh

加入jdk环境，并将hbase自带的zookeeper改为false，使用自己安装的zookeeper。

export JAVA_HOME=/usr/local/jdk1.8.0_211
export HBASE_MANAGES_ZK=false

修改hbase-site.xml

在中加入如下配置：

<property>
<!-- 指定 hbase 在 HDFS 上存储的路径 -->
<name>hbase.rootdir</name>
<value>hdfs://hadoop01:9000/hbase</value>
</property>
<property>
<!-- 指定 hbase 是分布式的 -->
<name>hbase.cluster.distributed</name>
<value>true</value>
</property>
<property>
<name>hbase.master</name>
<value>16000</value><!--这里注意了，只需端口即可，不必再写主机名称了！-->
</property>
<property>
<!-- 指定 zk 的地址，多个用“,”分割 -->
<name>hbase.zookeeper.quorum</name>
<value>hadoop01:2181,hadoop02:2181,hadoop03:2181</value>
</property>

修改regionservers

加入自己的regionservers服务器ip

hadoop02
hadoop03

将hbase分发给其他服务器

scp –r /usr/local/hbase-1.3.2 hadoop02:/usr/local
scp –r /usr/local/hbase-1.3.2 hadoop03:/usr/local

再在别的服务器上配置hbase的环境变量即可，配置完后，记得更新（source /etc/profile）。

启动测试

在hadoop01（namenode）上执行

start-hbase.sh  //启动命令
stop-hbase.sh   //停止命令

访问网址hadoop01:ip+16010可以看到hbase的启动情况，如下图所示:

hbase高可用

hbase默认开启高可用，开启备用服务器即可。

开启步骤如下：

1. Hadoop02启动备用节点

   hbase-daemon.sh start master

2. 登录web页面查看

   Hadoop01：16010

3. 杀死hadoop01号的主节点，

   去查看hadoop02的状态

HBase的使用

基本操作

1. 进入HBase客户端命令行

   hbase shell

2. 查看帮助

   hbase(main):001:0> help

3. 查看当前数据库中有哪些表

   hbase(main):002:0> list

表操作

1. 创建表

   hbase(main):003:0> create 'student','cf1','cf2'

2. 插入数据到表

   hbase(main):004:0> put 'student','1001','cf1:name','zhangsan'
   0 row(s) in 0.2770 seconds
   
   hbase(main):005:0> put 'student','1001','cf1:age','18'
   0 row(s) in 0.0150 seconds
   
   hbase(main):006:0> put 'student','1001','cf1:sex','M'
   0 row(s) in 0.0110 seconds

3. 扫描查看表数据

    scan 'student'
   ROW                               COLUMN+CELL
    1001                             column=cf1:age, timestamp=1595274595996, value=18
    1001                             column=cf1:name, timestamp=1595274557404, value=zhangsan
    1001                             column=cf1:sex, timestamp=1595274653044, value=M
    1002                             column=cf1:age, timestamp=1595274745713, value=20
    1002                             column=cf1:name, timestamp=1595274729513, value=lisi
    1002                             column=cf1:sex, timestamp=1595274762071, value=M
   2 row(s) in 0.1520 seconds

4. 查看表结构

   hbase(main):018:0> describe 'student'
   Table student is ENABLED
   student
   COLUMN FAMILIES DESCRIPTION
   {NAME => 'cf1', BLOOMFILTER => 'ROW', VERSIONS => '1', IN_MEMORY => 'false', KEEP_DELETED_CELLS => 'FALSE', DATA_BLOCK_ENCODING =>
   'NONE', TTL => 'FOREVER', COMPRESSION => 'NONE', MIN_VERSIONS => '0', BLOCKCACHE => 'true', BLOCKSIZE => '65536', REPLICATION_SCOPE
    => '0'}
   {NAME => 'cf2', BLOOMFILTER => 'ROW', VERSIONS => '1', IN_MEMORY => 'false', KEEP_DELETED_CELLS => 'FALSE', DATA_BLOCK_ENCODING =>
   'NONE', TTL => 'FOREVER', COMPRESSION => 'NONE', MIN_VERSIONS => '0', BLOCKCACHE => 'true', BLOCKSIZE => '65536', REPLICATION_SCOPE
    => '0'}
   2 row(s) in 0.0340 seconds

5. 更新指定字段的数据

   hbase(main):028:0> put 'student','1001','cf1:name','Make'
   0 row(s) in 0.0240 seconds
   //查看表和数据
   hbase(main):029:0> list
   TABLE
   student
   1 row(s) in 0.0090 seconds
   
   => ["student"]
   hbase(main):030:0> scan 'student'
   ROW                               COLUMN+CELL
    1001                             column=cf1:age, timestamp=1595274595996, value=18
    1001                             column=cf1:name, timestamp=1595277338218, value=Make
    1001                             column=cf1:sex, timestamp=1595274653044, value=M
    1002                             column=cf1:age, timestamp=1595274745713, value=20
    1002                             column=cf1:name, timestamp=1595274729513, value=lisi
    1002                             column=cf1:sex, timestamp=1595274762071, value=M
   2 row(s) in 0.0320 seconds

6. 查看“指定行”或“指定列族:列”的数据

   hbase(main):031:0> get 'student' ,'1001'
   COLUMN                            CELL
    cf1:age                          timestamp=1595274595996, value=18
    cf1:name                         timestamp=1595277338218, value=Make
    cf1:sex                          timestamp=1595274653044, value=M
   1 row(s) in 0.0420 seconds
   
   hbase(main):032:0> get 'student','1001','cf1:name'
   COLUMN                            CELL
    cf1:name                         timestamp=1595277338218, value=Make
   1 row(s) in 0.0150 seconds

7. 统计表数据行数

   hbase(main):033:0> count 'student'
   2 row(s) in 0.0250 seconds
   
   => 2

8. 删除数据

   删除某rowkey的全部数据：

   hbase(main):034:0> deleteall 'student','1001'

   删除某rowkey的某一列数据:

   hbase(main):035:0> delete 'student','1002','info:sex'

9. 清空表数据

   hbase(main):036:0> truncate 'student'

   注：清空表的操作顺序为先disable，然后再truncate。

10. 删除表

    首先需要先让该表为disable状态：

    hbase(main):037:0> disable 'student'

    然后才能drop这个表：

    hbase(main):038:0> drop 'student'

    提示：如果直接drop表，会报错：ERROR: Table student is enabled. Disable it first.

11. 变更表信息

    将info列族中的数据存放3个版本：

    hbase(main):039:0> alter 'student',{NAME=>'cf1',VERSIONS=>3}
    hbase(main):040:0> get 'student','1001',{COLUMN=>'cf1:name',VERSIONS=>3}

HBase数据结构

RowKey

与nosql数据库们一样,RowKey是用来检索记录的主键。访问HBASE table中的行，只有三种方式：

1. 通过单个RowKey访问

2. 通过RowKey的range（正则）

3. 全表扫描

RowKey行键 (RowKey)可以是任意字符串(最大长度是64KB，实际应用中长度一般为 10-100bytes)，在HBASE内部，RowKey保存为字节数组。存储时，数据按照RowKey的字典序(byte order)排序存储。设计RowKey时，要充分排序存储这个特性，将经常一起读取的行存储放到一起。(位置相关性)

Column Family

列族：HBASE表中的每个列，都归属于某个列族。列族是表的schema的一部 分(而列不是)，必须在使用表之前定义。列名都以列族作为前缀。例如 courses:history，courses:math都属于courses 这个列族。

Cell

由{rowkey, column Family:columu, version} 唯一确定的单元。cell中的数据是没有类型的，全部是字节码形式存贮。

关键字：无类型、字节码

Time Stamp

HBASE 中通过rowkey和columns确定的为一个存贮单元称为cell。每个 cell都保存 着同一份数据的多个版本。版本通过时间戳来索引。时间戳的类型是 64位整型。时间戳可以由HBASE(在数据写入时自动 )赋值，此时时间戳是精确到毫秒 的当前系统时间。时间戳也可以由客户显式赋值。如果应用程序要避免数据版 本冲突，就必须自己生成具有唯一性的时间戳。每个 cell中，不同版本的数据按照时间倒序排序，即最新的数据排在最前面。

为了避免数据存在过多版本造成的的管理 (包括存贮和索引)负担，HBASE提供 了两种数据版本回收方式。一是保存数据的最后n个版本，二是保存最近一段 时间内的版本（比如最近七天）。用户可以针对每个列族进行设置。

NameSpace(命名空间)

1) Table：表，所有的表都是命名空间的成员，即表必属于某个命名空间，如果没有指定，则在default默认的命名空间中。

2) RegionServer group：一个命名空间包含了默认的RegionServer Group。

3) Permission：权限，命名空间能够让我们来定义访问控制列表ACL（Access Control List）。例如，创建表，读取表，删除，更新等等操作。

4) Quota：限额，可以强制一个命名空间可包含的region的数量。

HBase读写流程

读流程

写流程

HBaseAIP

idea表操作：

HBase表操作
1.获取Configuration对象
2.判断表是否存在
3.创建表
4.向表中插入数据
5.获取表数据
  1>获取某一行数据
  2>获取某一行指定"列簇:列"的数据
  3>获取所有数据
6.删除多行数据
7.删除表

代码已上传至仓库，点击这里查看。

HBase的MapRdeuce操作

环境配置

1. 关闭集群

2. 配置hadoop-env.sh

[root@hadoop01 ~]# vi /usr/local/hadoop-2.7.3/etc/hadoop/hadoop-env.sh

将hbaselib下的jar包导入hadoop环境下。

3. 重启集群

   hadoop：start-all.sh

   zookeeper: zkstart.sh(自己封装的shell脚本，一键启动zookeeper集群)

   hbase：start-hbase.sh

4. 测试

   运行官方的MapReduce任务

   yarn jar lib/hbase-server-1.3.2.jar rowcounter student

   统计student表的行数(row)。

   

   student表内容如下：

   

MapReduce

在终端将本地数据导入到HBase。

1. 在本地创建一个tsv格式的文件：fruit.tsv

   1001	Apple	Red
   1002	Pear	Yellow
   1003	Pineapple	Yellow

2. 创建HBase表

   hbase(main):001:0> create 'fruit','info'

3. 在HDFS中创建input_fruit文件夹并上传fruit.tsv文件

   hadoop fs -mkdir -p /input_fruit/
   hadoop fs -put fruit.tsv /input_fruit/

4. 执行MapReduce到HBase的fruit表中

   yarn jar /usr/local/hbase-1.3.2/lib/hbase-server-1.3.2.jar importtsv \
   -Dimporttsv.columns=HBASE_ROW_KEY,info:name,info:color fruit \
   hdfs://hadoop01:9000/input_fruit

5. 使用scan命令查看导入后的结果

   hbase(main):001:0> scan ‘fruit’

自定义MapReduce（Idea）

将Hbase_farm.tsv表中的一部分数据，通过MR迁入到farm_mr表中。

1. 在本地创建一个tsv格式的文件：将farm1.tsv数据转变成Hbase_farm01.tsv（rowkey设计）

   在所有字段前添加自增主键（即rowkey）

   #farm1.tsv
   香菜    2.80     2020-01-01     山西汾阳市晋阳农副产品批发市场  山西    汾阳
   #Hbase_farm.tsv
   1       香菜    2.80     2020-01-01     山西汾阳市晋阳农副产品批发市场  山西    汾阳

   使用awk命令为数据添加rowkey值

   awk -F '\t' 'BEGIN{s=0}{s++} {print s"\t"$1"\t"$2"\t"$3"\t"$4"\t"$5"\t"$6}' farm1.tsv >> Hbase_farm01.tsv

2. 创建HBase表

   hbase(main):001:0> create 'fruit','info'

3. 在HDFS中创建input_fruit文件夹并上传fruit.tsv文件

   hadoop fs -mkdir -p /input_fruit/
   hadoop fs -put fruit.tsv /input_fruit/

4. 执行MapReduce到HBase的fruit表中

   package com.lcx.hbaseMapReduce;
   
   import org.apache.hadoop.conf.Configuration;
   import org.apache.hadoop.fs.Path;
   import org.apache.hadoop.hbase.HBaseConfiguration;
   import org.apache.hadoop.hbase.client.Put;
   import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
   import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
   import org.apache.hadoop.hbase.mapreduce.TableReducer;
   import org.apache.hadoop.hbase.util.Bytes;
   import org.apache.hadoop.io.LongWritable;
   import org.apache.hadoop.io.NullWritable;
   import org.apache.hadoop.io.Text;
   import org.apache.hadoop.mapreduce.Job;
   import org.apache.hadoop.mapreduce.Mapper;
   import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
   
   import java.io.IOException;
   
   //给文件加自增id
   //awk -F '\t' 'BEGIN{s=0}{s++} {print s"\t"$1"\t"$2"\t"$3"\t"$4"\t"$5"\t"$6}' farm1.txv >> farm2.txv
   public class MapReduce02 {
   
       //map
       public static class MapTask extends Mapper<LongWritable, Text, ImmutableBytesWritable, Put> {
           @Override
           protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
               String[] split = value.toString().split("\t");
               if (split.length >= 7) {
                   Put put = new Put(Bytes.toBytes(split[0]));
                   ImmutableBytesWritable rowkey = new ImmutableBytesWritable(Bytes.toBytes(split[0]));
                   put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("name"), Bytes.toBytes(split[1]));
                   put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("price"), Bytes.toBytes(split[2]));
                   put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("times"), Bytes.toBytes(split[3]));
                   put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("market"), Bytes.toBytes(split[4]));
                   put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("province"), Bytes.toBytes(split[5]));
                   put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("city"), Bytes.toBytes(split[6]));
                   context.write(rowkey, put);
               }
           }
       }
   
       //reduce
       public static class HbaseReduceTask extends TableReducer<ImmutableBytesWritable, Put, NullWritable> {
           @Override
           protected void reduce(ImmutableBytesWritable key, Iterable<Put> values, Context context) throws IOException, InterruptedException {
               for (Put value : values) {
                   context.write(NullWritable.get(), value);
               }
           }
       }
   
       //main
       public static void main(String[] args) throws Exception {
           Configuration conf = HBaseConfiguration.create();
           conf.set("hbase.zookeeper.quorum", "hadoop01");
           conf.set("hbase.zookeeper.property.clientPort", "2181");
   
           Job job = Job.getInstance(conf);
   
           job.setMapOutputKeyClass(ImmutableBytesWritable.class);
           job.setMapOutputValueClass(Put.class);
           job.setMapperClass(MapTask.class);
           job.setJarByClass(MapReduce02.class);
   
           //hdfs 输入输出路径
           FileInputFormat.addInputPath(job, new Path("hdfs://hadoop01:9000/input_farm/Hbase_farm01.tsv"));
   
           TableMapReduceUtil.initTableReducerJob("farm_mr", HbaseReduceTask.class, job);
   
           //温馨提示
           boolean b = job.waitForCompletion(true);
           System.out.println(b ? true : false);
       }
   }

5. 使用scan命令查看导入后的结果

   hbase(main):001:0> scan ‘farm_mr’

HBase与Hive集成

HBase与Hive的对比

1．Hive

(1) 数据仓库

Hive的本质其实就相当于将HDFS中已经存储的文件在Mysql中做了一个双射关系，以方便使用HQL去管理查询。

(2) 用于数据分析、清洗

Hive适用于离线的数据分析和清洗，延迟较高。

(3) 基于HDFS、MapReduce

Hive存储的数据依旧在DataNode上，编写的HQL语句终将是转换为MapReduce代码执行。

2．HBase

(1) 数据库

是一种面向列存储的非关系型数据库。

(2) 用于存储结构化和非结构化的数据

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

(3) 基于HDFS

数据持久化存储的体现形式是Hfile，存放于DataNode中，被ResionServer以region的形式进行管理。

(4) 延迟较低，接入在线业务使用

面对大量的企业数据，HBase可以直线单表大量数据的存储，同时提供了高效的数据访问速度。

HBase与Hive集成使用

尖叫提示：HBase与Hive的集成在最新的两个版本中无法兼容。所以，我们只能含着泪勇敢的重新编译：hive-hbase-handler-1.2.2.jar！！好气！！

从Hive中自动导入数据到HBase表中

3.hbase与hive集成

//创建关联表
CREATE TABLE hive_hbase_farm(
id int,
name string,
price double,
times string,
market string,
province string,
city string)
STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'
WITH SERDEPROPERTIES ("hbase.columns.mapping" = ":key,cf1:name,cf1:times,cf1:times,cf1:price,cf1:province,cf1:city")
TBLPROPERTIES ("hbase.table.name" = "hbase_farm_table");
//创建临时表
CREATE TABLE emp(
id int,
name string,
price double,
times string,
market string,
province string,
city string)
row format delimited fields terminated by '\t';
//给临时表中导入数据
load data local inpath '/root/Hbase_farm01.tsv' into table emp;

//导入hbase
insert into table hive_hbase_farm select * from emp;

从HBase表中导入数据至Hive中

在实际生产中，hive从hbase数据仓库中拉去数据，在hive中通过类sql语句进行mapreduce操作，对数据进行处理。

将HBase中的stu2表中的数据导入hive的hbase_stu2表中：

//将hbase导入hive（更接近于真实场景）
//hive中建外部表
CREATE EXTERNAL TABLE hbase_stu2(
id int,
name string,
age int)
STORED BY 
'org.apache.hadoop.hive.hbase.HBaseStorageHandler'
WITH SERDEPROPERTIES ("hbase.columns.mapping" = 
":key,info:name,info:age") 
TBLPROPERTIES ("hbase.table.name" = "stu2");

stu2表中数据如下：

hbase_stu2表中的结果如下：s