大数据之Hadoop分布式数据仓库HBase_hadoop hbase

一、Hadoop的局限

HBase 是一个构建在 Hadoop 文件系统之上的面向列的数据库管理系统。

要想明白为什么产生 HBase，就需要先了解一下 Hadoop 存在的限制？Hadoop 可以通过 HDFS 来存储结构化、半结构甚至非结构化的数据，它是传统数据库的补充，是海量数据存储的最佳方法，它针对大文件的存储，批量访问和流式访问都做了优化，同时也通过多副本解决了容灾问题。

但是 Hadoop 的缺陷在于它只能执行批处理，并且只能以顺序方式访问数据，这意味着即使是最简单的工作，也必须搜索整个数据集，无法实现对数据的随机访问。实现数据的随机访问是传统的关系型数据库所擅长的，但它们却不能用于海量数据的存储。在这种情况下，必须有一种新的方案来解决海量数据存储和随机访问的问题，HBase 就是其中之一 (HBase，Cassandra，couchDB，Dynamo 和 MongoDB 都能存储海量数据并支持随机访问)。

注：数据结构分类：

• 结构化数据：即以关系型数据库表形式管理的数据；
• 半结构化数据：非关系模型的，有基本固定结构模式的数据，例如日志文件、XML 文档、JSON文档、Email 等；
• 非结构化数据：没有固定模式的数据，如 WORD、PDF、PPT、EXL，各种格式的图片、视频等。

二、HBase简介

HBase 是一个构建在 Hadoop 文件系统之上的面向列的数据库管理系统。

HBase 是一种类似于 Google’s Big Table 的数据模型，它是 Hadoop 生态系统的一部分，它将数据存储在 HDFS 上，客户端可以通过 HBase 实现对 HDFS 上数据的随机访问。它具有以下特性：

• 不支持复杂的事务，只支持行级事务，即单行数据的读写都是原子性的；
• 由于是采用 HDFS 作为底层存储，所以和 HDFS 一样，支持结构化、半结构化和非结构化的存储；
• 支持通过增加机器进行横向扩展；
• 支持数据分片；
• 支持 RegionServers 之间的自动故障转移；
• 易于使用的 Java 客户端 API；
• 支持 BlockCache 和布隆过滤器；
• 过滤器支持谓词下推。

三、HBase Table

HBase 是一个面向 列 的数据库管理系统，这里更为确切的而说，HBase 是一个面向 列族 的数据库管理系统。表 schema 仅定义列族，表具有多个列族，每个列族可以包含任意数量的列，列由多个单元格（cell ）组成，单元格可以存储多个版本的数据，多个版本数据以时间戳进行区分。

下图为 HBase 中一张表的：

• RowKey 为行的唯一标识，所有行按照 RowKey 的字典序进行排序；
• 该表具有两个列族，分别是 personal 和 office;
• 其中列族 personal 拥有 name、city、phone 三个列，列族 office 拥有 tel、addres 两个列。
  
  Hbase 的表具有以下特点：
• 容量大：一个表可以有数十亿行，上百万列；
• 面向列：数据是按照列存储，每一列都单独存放，数据即索引，在查询时可以只访问指定列的数据，有效地降低了系统的 I/O 负担；
• 稀疏性：空 (null) 列并不占用存储空间，表可以设计的非常稀疏 ；
• 数据多版本：每个单元中的数据可以有多个版本，按照时间戳排序，新的数据在最上面；
• 存储类型：所有数据的底层存储格式都是字节数组 (byte[])。

四、数据的读写流程简述

4.1 写入数据的流程

• Client 向 Region Server 提交写请求；
• Region Server 找到目标 Region；
• Region 检查数据是否与 Schema 一致；
• 如果客户端没有指定版本，则获取当前系统时间作为数据版本；
• 将更新写入 WAL Log；
• 将更新写入 Memstore；
• 判断 Memstore 存储是否已满，如果存储已满则需要 flush 为 Store Hfile 文件。

4.2 读取数据的流程

以下是客户端首次读写 HBase 上数据的流程：

• 客户端从 Zookeeper 获取 META 表所在的 Region Server；
• 客户端访问 META 表所在的 Region Server，从 META 表中查询到访问行键所在的 Region
  Server，之后客户端将缓存这些信息以及 META 表的位置；
• 客户端从行键所在的 Region Server 上获取数据。

如果再次读取，客户端将从缓存中获取行键所在的 Region Server。这样客户端就不需要再次查询 META 表，除非 Region 移动导致缓存失效，这样的话，则将会重新查询并更新缓存。

注：META 表是 HBase 中一张特殊的表，它保存了所有 Region 的位置信息，META 表自己的位置信息则存储在 ZooKeeper 上。

五、HBase Java API 1.0 的基本使用

2.1 新建Maven工程，导入项目依赖
要使用 Java API 操作 HBase，需要引入 hbase-client。这里选取的 HBase Client 的版本为 1.2.0。

<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-client</artifactId>
    <version>1.2.0</version>
</dependency>
1
2
3
4
5

2.3 单元测试

以单元测试的方式对上面封装的 API 进行测试（工具类参考我的工具类专栏）。

public class HBaseUtilsTest {

    private static final String TABLE_NAME = "class";
    private static final String TEACHER = "teacher";
    private static final String STUDENT = "student";

    @Test
    public void createTable() {
        // 新建表
        List<String> columnFamilies = Arrays.asList(TEACHER, STUDENT);
        boolean table = HBaseUtils.createTable(TABLE_NAME, columnFamilies);
        System.out.println("表创建结果:" + table);
    }

    @Test
    public void insertData() {
        List<Pair<String, String>> pairs1 = Arrays.asList(new Pair<>("name", "Tom"),
                new Pair<>("age", "22"),
                new Pair<>("gender", "1"));
        HBaseUtils.putRow(TABLE_NAME, "rowKey1", STUDENT, pairs1);

        List<Pair<String, String>> pairs2 = Arrays.asList(new Pair<>("name", "Jack"),
                new Pair<>("age", "33"),
                new Pair<>("gender", "2"));
        HBaseUtils.putRow(TABLE_NAME, "rowKey2", STUDENT, pairs2);

        List<Pair<String, String>> pairs3 = Arrays.asList(new Pair<>("name", "Mike"),
                new Pair<>("age", "44"),
                new Pair<>("gender", "1"));
        HBaseUtils.putRow(TABLE_NAME, "rowKey3", STUDENT, pairs3);
    }


    @Test
    public void getRow() {
        Result result = HBaseUtils.getRow(TABLE_NAME, "rowKey1");
        if (result != null) {
            System.out.println(Bytes
                    .toString(result.getValue(Bytes.toBytes(STUDENT), Bytes.toBytes("name"))));
        }

    }

    @Test
    public void getCell() {
        String cell = HBaseUtils.getCell(TABLE_NAME, "rowKey2", STUDENT, "age");
        System.out.println("cell age :" + cell);

    }

    @Test
    public void getScanner() {
        ResultScanner scanner = HBaseUtils.getScanner(TABLE_NAME);
        if (scanner != null) {
            scanner.forEach(result -> System.out.println(Bytes.toString(result.getRow()) + "->" + Bytes
                    .toString(result.getValue(Bytes.toBytes(STUDENT), Bytes.toBytes("name")))));
            scanner.close();
        }
    }


    @Test
    public void getScannerWithFilter() {
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
        SingleColumnValueFilter nameFilter = new SingleColumnValueFilter(Bytes.toBytes(STUDENT),
                Bytes.toBytes("name"), CompareOperator.EQUAL, Bytes.toBytes("Jack"));
        filterList.addFilter(nameFilter);
        ResultScanner scanner = HBaseUtils.getScanner(TABLE_NAME, filterList);
        if (scanner != null) {
            scanner.forEach(result -> System.out.println(Bytes.toString(result.getRow()) + "->" + Bytes
                    .toString(result.getValue(Bytes.toBytes(STUDENT), Bytes.toBytes("name")))));
            scanner.close();
        }
    }

    @Test
    public void deleteColumn() {
        boolean b = HBaseUtils.deleteColumn(TABLE_NAME, "rowKey2", STUDENT, "age");
        System.out.println("删除结果: " + b);
    }

    @Test
    public void deleteRow() {
        boolean b = HBaseUtils.deleteRow(TABLE_NAME, "rowKey2");
        System.out.println("删除结果: " + b);
    }

    @Test
    public void deleteTable() {
        boolean b = HBaseUtils.deleteTable(TABLE_NAME);
        System.out.println("删除结果: " + b);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93

六、HBase Java API 2.0 的基本使用

3.1 新建Maven工程，导入项目依赖

这里选取的 HBase Client 的版本为最新的 2.1.4。

<dependency>
    <groupId>org.apache.hbase</groupId>
    <artifactId>hbase-client</artifactId>
    <version>2.1.4</version>
</dependency>
1
2
3
4
5

工具类的使用参考我的工具类专栏

七、正确连接Hbase

在上面的代码中，在类加载时就初始化了 Connection 连接，并且之后的方法都是复用这个 Connection，这时我们可能会考虑是否可以使用自定义连接池来获取更好的性能表现？实际上这是没有必要的。

HBase 客户端需要连接三个不同的服务角色：

• Zookeeper ：主要用于获取 meta 表的位置信息，Master 的信息；
• HBase Master ：主要用于执行 HBaseAdmin 接口的一些操作，例如建表等；
• HBase RegionServer ：用于读、写数据。
  

HBase 中提供了三种资源池的实现，分别是 Reusable，RoundRobin 和 ThreadLocal。具体实现可以通 hbase.client.ipc.pool.type 配置项指定，默认为 Reusable。连接池的大小也可以通过 hbase.client.ipc.pool.size 配置项指定，默认为 1，即每个 Server 1 个连接。也可以通过修改配置实现：

config.set("hbase.client.ipc.pool.type",...);
config.set("hbase.client.ipc.pool.size",...);
connection = ConnectionFactory.createConnection(config);
1
2
3

八、Hbase 常用 Shell 命令

一、基本命令
打开 Hbase Shell：
1
2

# hbase shell
1.1 获取帮助
# 获取帮助
help
1
2
3
4

# 获取命令的详细信息
help 'status'
1.2 查看服务器状态
status
1.3 查看版本信息
version
二、关于表的操作
2.1 查看所有表
list
2.2 创建表
命令格式： create '表名称', '列族名称 1','列族名称 2','列名称 N'
1
2
3
4
5
6
7
8
9
10
11

# 创建一张名为Student的表,包含基本信息（baseInfo）、学校信息（schoolInfo）两个列族
create 'Student','baseInfo','schoolInfo'
2.3 查看表的基本信息
命令格式：desc '表名'

describe 'Student'
2.4 表的启用/禁用
enable 和 disable 可以启用/禁用这个表,is_enabled 和 is_disabled 来检查表是否被禁用
1
2
3
4
5
6
7
8

# 禁用表
disable 'Student'
1
2

# 检查表是否被禁用
is_disabled 'Student'
1
2

# 启用表
enable 'Student'
1
2

# 检查表是否被启用
is_enabled 'Student'
1
2

2.5 检查表是否存在
exists 'Student'
2.6 删除表
# 删除表前需要先禁用表
disable 'Student'
# 删除表
drop 'Student'
三、增删改
3.1 添加列族
命令格式： alter '表名', '列族名'

alter 'Student', 'teacherInfo'
3.2 删除列族
命令格式：alter '表名', {NAME => '列族名', METHOD => 'delete'}

alter 'Student', {NAME => 'teacherInfo', METHOD => 'delete'}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

```powershell
3.3 更改列族存储版本的限制
默认情况下，列族只存储一个版本的数据，如果需要存储多个版本的数据，则需要修改列族的属性。修改后可通过 desc 命令查看。

alter 'Student',{NAME=>'baseInfo',VERSIONS=>3}
1
2
3
4
5
6

3.4 插入数据
命令格式：put ‘表名’, ‘行键’,‘列族:列’,‘值’

注意：如果新增数据的行键值、列族名、列名与原有数据完全相同，则相当于更新操作

put ‘Student’, ‘rowkey1’,‘baseInfo:name’,‘tom’
put ‘Student’, ‘rowkey1’,‘baseInfo:birthday’,‘1990-01-09’
put ‘Student’, ‘rowkey1’,‘baseInfo:age’,‘29’
put ‘Student’, ‘rowkey1’,‘schoolInfo:name’,‘Havard’
put ‘Student’, ‘rowkey1’,‘schoolInfo:localtion’,‘Boston’

put ‘Student’, ‘rowkey2’,‘baseInfo:name’,‘jack’
put ‘Student’, ‘rowkey2’,‘baseInfo:birthday’,‘1998-08-22’
put ‘Student’, ‘rowkey2’,‘baseInfo:age’,‘21’
put ‘Student’, ‘rowkey2’,‘schoolInfo:name’,‘yale’
put ‘Student’, ‘rowkey2’,‘schoolInfo:localtion’,‘New Haven’

put ‘Student’, ‘rowkey3’,‘baseInfo:name’,‘maike’
put ‘Student’, ‘rowkey3’,‘baseInfo:birthday’,‘1995-01-22’
put ‘Student’, ‘rowkey3’,‘baseInfo:age’,‘24’
put ‘Student’, ‘rowkey3’,‘schoolInfo:name’,‘yale’
put ‘Student’, ‘rowkey3’,‘schoolInfo:localtion’,‘New Haven’

put ‘Student’, ‘wrowkey4’,‘baseInfo:name’,‘maike-jack’

3.5 获取指定行、指定行中的列族、列的信息

```powershell
# 获取指定行中所有列的数据信息
get 'Student','rowkey3'
# 获取指定行中指定列族下所有列的数据信息
get 'Student','rowkey3','baseInfo'
# 获取指定行中指定列的数据信息
get 'Student','rowkey3','baseInfo:name'
1
2
3
4
5
6
7
8
9
10

3.6 删除指定行、指定行中的列

# 删除指定行
delete 'Student','rowkey3'
# 删除指定行中指定列的数据
delete 'Student','rowkey3','baseInfo:name'
1
2
3
4

四、查询
hbase 中访问数据有两种基本的方式：

• 按指定 rowkey 获取数据：get 方法；
• 按指定条件获取数据：scan 方法。

scan 可以设置 begin 和 end 参数来访问一个范围内所有的数据。get 本质上就是 begin 和 end 相等的一种特殊的 scan。

4.1Get查询

# 获取指定行中所有列的数据信息
get 'Student','rowkey3'
1
2

# 获取指定行中指定列族下所有列的数据信息
get 'Student','rowkey3','baseInfo'
1
2

# 获取指定行中指定列的数据信息
get 'Student','rowkey3','baseInfo:name'
1
2

4.2 查询整表数据
scan 'Student'
4.3 查询指定列簇的数据
scan 'Student', {COLUMN=>'baseInfo'}
4.4 条件查询
# 查询指定列的数据
scan 'Student', {COLUMNS=> 'baseInfo:birthday'}
除了列 （COLUMNS） 修饰词外，HBase 还支持 Limit（限制查询结果行数），STARTROW（ROWKEY 起始行，会先根据这个 key 定位到 region，再向后扫描）、STOPROW(结束行)、TIMERANGE（限定时间戳范围）、VERSIONS（版本数）、和 FILTER（按条件过滤行）等。

如下代表从 rowkey2 这个 rowkey 开始，查找下两个行的最新 3 个版本的 name 列的数据：

scan 'Student', {COLUMNS=> 'baseInfo:name',STARTROW => 'rowkey2',STOPROW => 'wrowkey4',LIMIT=>2, VERSIONS=>3}
1
2
3
4
5
6
7
8
9
10
11
12

4.5 条件过滤

Filter 可以设定一系列条件来进行过滤。如我们要查询值等于 24 的所有数据：

scan 'Student', FILTER=>"ValueFilter(=,'binary:24')"
值包含 yale 的所有数据：

scan 'Student', FILTER=>"ValueFilter(=,'substring:yale')"
列名中的前缀为 birth 的：

scan 'Student', FILTER=>"ColumnPrefixFilter('birth')"
FILTER 中支持多个过滤条件通过括号、AND 和 OR 进行组合：

# 列名中的前缀为birth且列值中包含1998的数据
scan 'Student', FILTER=>"ColumnPrefixFilter('birth') AND ValueFilter ValueFilter(=,'substring:1998')"
PrefixFilter 用于对 Rowkey 的前缀进行判断：

scan 'Student', FILTER=>"PrefixFilter('wr')"
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

九、Hbase容灾与备份

一、前言
本文主要介绍 Hbase 常用的三种简单的容灾备份方案，即CopyTable、Export/Import、Snapshot。分别介绍如下：

二、CopyTable
2.1 简介
CopyTable可以将现有表的数据复制到新表中，具有以下特点：

支持时间区间 、row 区间 、改变表名称 、改变列族名称 、以及是否 Copy 已被删除的数据等功能；
执行命令前，需先创建与原表结构相同的新表；
CopyTable 的操作是基于 HBase Client API 进行的，即采用 scan 进行查询, 采用 put 进行写入。

2.2 命令格式

Usage: CopyTable [general options] [--starttime=X] [--endtime=Y] [--new.name=NEW] [--peer.adr=ADR] <tablename>
1

2.3 常用命令
同集群下 CopyTable

hbase org.apache.hadoop.hbase.mapreduce.CopyTable --new.name=tableCopy  tableOrig
1

不同集群下 CopyTable

# 两表名称相同的情况
hbase org.apache.hadoop.hbase.mapreduce.CopyTable \
--peer.adr=dstClusterZK:2181:/hbase tableOrig
1
2
3

# 也可以指新的表名
hbase org.apache.hadoop.hbase.mapreduce.CopyTable \
--peer.adr=dstClusterZK:2181:/hbase \
--new.name=tableCopy tableOrig
下面是一个官方给的比较完整的例子，指定开始和结束时间，集群地址，以及只复制指定的列族：
hbase org.apache.hadoop.hbase.mapreduce.CopyTable \
--starttime=1265875194289 \
--endtime=1265878794289 \
--peer.adr=server1,server2,server3:2181:/hbase \
--families=myOldCf:myNewCf,cf2,cf3 TestTable
1
2
3
4
5
6
7
8
9
10

2.4 更多参数
可以通过 --help 查看更多支持的参数

# hbase org.apache.hadoop.hbase.mapreduce.CopyTable --help
1

三、Export/Import
3.1 简介
Export 支持导出数据到 HDFS, Import 支持从 HDFS 导入数据。Export 还支持指定导出数据的开始时间和结束时间，因此可以用于增量备份。
Export 导出与 CopyTable 一样，依赖 HBase 的 scan 操作

3.2 命令格式

# Export
hbase org.apache.hadoop.hbase.mapreduce.Export <tablename> <outputdir> [<versions> [<starttime> [<endtime>]]]
1
2

# Inport
hbase org.apache.hadoop.hbase.mapreduce.Import <tablename> <inputdir>
导出的 outputdir 目录可以不用预先创建，程序会自动创建。导出完成后，导出文件的所有权将由执行导出命令的用户所拥有。
默认情况下，仅导出给定 Cell 的最新版本，而不管历史版本。要导出多个版本，需要将 <versions> 参数替换为所需的版本数。
1
2
3
4

3.3 常用命令

导出命令
hbase org.apache.hadoop.hbase.mapreduce.Export tableName  hdfs 路径/tableName.db
1
2

导入命令
hbase org.apache.hadoop.hbase.mapreduce.Import tableName  hdfs 路径/tableName.db
1
2

四、Snapshot
4.1 简介
HBase 的快照 (Snapshot) 功能允许您获取表的副本 (包括内容和元数据)，并且性能开销很小。因为快照存储的仅仅是表的元数据和 HFiles 的信息。快照的 clone 操作会从该快照创建新表，快照的 restore 操作会将表的内容还原到快照节点。clone 和 restore 操作不需要复制任何数据，因为底层 HFiles(包含 HBase 表数据的文件) 不会被修改，修改的只是表的元数据信息。

4.2 配置
HBase 快照功能默认没有开启，如果要开启快照，需要在 hbase-site.xml 文件中添加如下配置项：

<property>
    <name>hbase.snapshot.enabled</name>
    <value>true</value>
</property>
1
2
3
4

4.3 常用命令
快照的所有命令都需要在 Hbase Shell 交互式命令行中执行。

1. Take a Snapshot

# 拍摄快照
hbase> snapshot '表名', '快照名'
1
2

默认情况下拍摄快照之前会在内存中执行数据刷新。以保证内存中的数据包含在快照中。但是如果你不希望包含内存中的数据，则可以使用 SKIP_FLUSH 选项禁止刷新。

# 禁止内存刷新
hbase> snapshot  '表名', '快照名', {SKIP_FLUSH => true}
1
2

2. Listing Snapshots

# 获取快照列表
hbase> list_snapshots
1
2

3. Deleting Snapshots

# 删除快照
hbase> delete_snapshot '快照名'
1
2

4. Clone a table from snapshot

# 从现有的快照创建一张新表
hbase>  clone_snapshot '快照名', '新表名'
1
2

5. Restore a snapshot
   将表恢复到快照节点，恢复操作需要先禁用表

hbase> disable '表名'
hbase> restore_snapshot '快照名'
1
2

这里需要注意的是：是如果 HBase 配置了基于 Replication 的主从复制，由于 Replication 在日志级别工作，而快照在文件系统级别工作，因此在还原之后，会出现副本与主服务器处于不同的状态的情况。这时候可以先停止同步，所有服务器还原到一致的数据点后再重新建立同步。

十、引入Phoenix core JAR包（HBase中间层，可以使用jdbc操作数据库）

如果是 maven 项目，直接在 maven 中央仓库找到对应的版本，导入依赖即可：

 <!-- https://mvnrepository.com/artifact/org.apache.phoenix/phoenix-core -->
    <dependency>
      <groupId>org.apache.phoenix</groupId>
      <artifactId>phoenix-core</artifactId>
      <version>4.14.0-cdh5.14.2</version>
    </dependency>
1
2
3
4
5
6

如果是普通项目，则可以从 Phoenix 解压目录下找到对应的 JAR 包，然后手动引入：

代码示例：

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;


public class PhoenixJavaApi {

    public static void main(String[] args) throws Exception {

        // 加载数据库驱动
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");

        /*
         * 指定数据库地址,格式为 jdbc:phoenix:Zookeeper 地址
         * 如果 HBase 采用 Standalone 模式或者伪集群模式搭建，则 HBase 默认使用内置的 Zookeeper，默认端口为 2181
         */
        Connection connection = DriverManager.getConnection("jdbc:phoenix:192.168.200.226:2181");

        PreparedStatement statement = connection.prepareStatement("SELECT * FROM us_population");

        ResultSet resultSet = statement.executeQuery();

        while (resultSet.next()) {
            System.out.println(resultSet.getString("city") + " "
                    + resultSet.getInt("population"));
        }

        statement.close();
        connection.close();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

整合mybatis来操作