Hadoop 综合揭秘——HBase的原理与应用 - 风尘浪子

时间：2024-01-21 15:00:18

Hadoop 综合揭秘——HBase的原理与应用

前言

现今互联网科技发展日新月异，大数据、云计算、人工智能等技术已经成为前瞻性产品，海量数据和超高并发让传统的 Web2.0 网站有点力不从心，暴露了很多难以克服的问题。为此，Google、Amazon 、Powerset 等各大平台纷纷推出 NoSQL 技术以应对市场的急速发展，近10年间NoSQL技术百花齐放，HBase、Redis、MongoDB、Cassandra 等技术纷纷涌现。
本文主要向各位介绍 HBase 的发展历史，基础结构与原理，应用的场景，对常用的 JAVA API 操作进行梳理，在最后一节还会详细讲述 HBase 与 MR 之间关系。

 

 

目录

一、HBase 的概述

二、HBase 的原理

三、简述 RowKey 设计原理

四、HBase 的 Java API 开发实例

五、HBase与MapReduce的相互调用

 

 

 

一、HBase 的概述

1.1 HBase 的发展历史

HBase（Hadoop Database）是一个高可靠性、高性能、面向列、可伸缩的分布式数据库，典型的 NoSQL（Not Only SQL）数据库。它起源于 Hadoop 的子项目，由 Powerset 公司在2007年创建，同年10 月 HBase 的第一版与 Hadoop 0.15.0 捆绑发布，初期的目标是弥补 MapReduce 在实时操作上的缺失，方便用户可随时操作大规模的数据集。随着大数据与 NoSQL 的流行和迅速发展，在 2010年5月，Apache HBase 脱离了 Hadoop，成为 Apache 基金的*项目。次年即 2011年1月 ZooKeeper 也脱离 Hadoop，成为 Apache 基金的*项目。

1.2 HBase 的特点

  • 面向列设计：面向列表（簇）的存储和权限控制，列（簇）独立检索。
  • 支持多版本:每个单元中的数据可以有多个版本，默认情况下，版本号可自动分配，版本号就是单元格插入时的时间戳。
  • 稀疏性：为空的列不占用存储空间，表可以设计得非常稀疏。
  • 高可靠性：WAL机制保证了数据写入时不会因集群异常而导致写入数据丢失，Replication机制保证了在集群出现严重的问题时，数据不会发生丢失或损坏。
  • 高性能：底层的LSM数据结构和 Rowkey 有序排列等架构上的独特设计，使得Hbase具有非常高的写入性能。通过科学性地设计RowKey 可让数据进行合理的 Region 切分，主键索引和缓存机制使得Hbase 在海量数据下具备高速的随机读取性能。（下文将再作介绍）

1.3 HBase （NoSQL）与 RDBMS 的区别

1.3.1 传统的 RDBMS 具有以下特征

它是面向表格、视图设计的标准化数据，表中的数据类型也会进行预定义，数据保存后表的结构不易修改。每个表格对列的数据有所限制，最大不会超过几个百个，这将导致不同的数据可能会存放到多个表，表格之间存在一对一，一对多，多对一，多对多等复杂关系。正因如此也限制了 RDBMS 的使用场景更适合于高度结构化的的行业，例如医疗，机关，教育等行业。

1.3.2 HBase 是典型的 NoSQL代表

相对于 RDBMS ，它属于一种高效的映射嵌套型弱视图设计，以Key-Value的方式存储数据，每一行数据都可以有不同的列设计。数据依赖于行键作为唯一标识，当行数据的结构发生变生时，HBase 也能根据需求作出灵活调整。数据以文本方式保存，HBase 把数据的解释任务交给了应用程序，因此它更适合于灵活的数据结构项目。

1.4 HBase 的版本问题

HBase 对 Hadoop 和 JDK 的版本支持性有一定要求，详细内容可在官网查询 http://hbase.apache.org/book.html

Hadoop version support matrix
"S" =supported ，"X"= not supported ，“NT"=Not tested

返回目录

二、HBase 的原理

2.1 HBase 的总体结构

HBase 的架构是依托于 Hadoop 的 HDFS 作为最基本存储基础单元，在 HBase 的集群中由一个 Master 主节点管理多个 Region Server ，而 Zookeeper 进行协调操作，其关系如下图所示：

2.1.1 HMaster

HMaster 用于启动任务管理多个HRegionServer，侦测各个HRegionServer之间的状态，当一个新的HRegionServer登录到HMaster时，HMaster 会告诉它等待分配数据，平衡 HRegionServer 之间的负载。而当某个 HRegionServer 死机时，HMaster会把它负责的所有HRegion标记为未分配，然后再把它们分配到其他 HRegionServer 中，并恢复HRegionServer的故障。事实上 HMaster 的负载很轻， HBase 允许有多个 HMaster 节点共存，但同一时刻只有一个 HMaster 能为系统提供服务，其他的 HMaster 节点处于待命的状态。当正在工作的 HMaster 节点宕机时，其他的 HMaster 则会接管HBase的集群。

2.1.2 HRegionServer

HBase中的所有数据从底层来说一般都是保存在HDFS中的，用户通过一系列HRegionServer获取这些数据。集群一个节点上一般只运行一个HRegionServer，且每一个区段的HRegion只会被一个HRegionServer维护。HRegionServer主要负责响应用户I/O请求，向HDFS文件系统读写数据，是HBase中最核心的模块。

2.1.3 Zookeeper

Apache Zookeeper 起源于 Hadoop 的分布式协同服务，是负责协调集群中的分布式组件，在 2011年1月 ZooKeeper 脱离 Hadoop，成为 Apache 基金的*项目。经过多年的发展 Zookeeper 已经成为了分布式大数据框架中容错性的标准框架，被多个分布式开源框架所应用。HBase 的组件之间是通过心跳机制协调系统之间的状态和健康信息的，这些功能都是通过消息实现，一旦消息因外界原因丢失，系统侧需要根据不同的情况进行处理， Zookeeper 的主要作用正是监听并协调各组件的运作。它监听了多个节点的使用状态，保证了 HMaster 处于正常运行当中，一旦 HMaster 发生故障时 Zookeeper 就会发出通知，备用的 HMaster 就会进行替代。Zookeeper 也会监测 HRegionServer 的健康状态， 一旦发生故障就会通知 HMaster ，把任务重新分配给正常的 HRegionServer 进行操作，并恢复有故障的 HRegionServer。

2.2 HBase 运作原理

在介绍完 HBase 的总体结构后，下面将为大家介绍一下 HRegion、HStore、MemStore、HFile、WAL 等组件是如何进行协调操作的，HBase 的运作原理图如下：

 

2.2.1 HRegion

每个 HRegionServer 内部管理了一系列 HRegion ，他们可以分别属于不同的逻辑表，每个 HRegion 对应了逻辑表中的一个连续数据段。HRegionServer 只是管理表格，实现读写操作。Client 直接连接到 HRegionServer，并通信获取 HBase 中的数据。而 HRegion 则是真实存放 HBase 数据的地方，也就说 HRegion 是 HBase 可用性和分布式的基本单位。当表的大小超过预设值的时候，HBase会自动将表划分为不同的区域，每个区域就是一个HRegion，以主键（RowKey）来区分。一个HRegion会保存一个表中某段连续的数据，一张完整的表数据是保存在多个 HRegion 中的，这些 HRegion 可以在同一个HRegionServer 中，也可以来源于不同的 HRegionServer。

2.2.2 HStore

每个 HRegion 由多个HStore组成，每个 HStore 对应逻辑表在这个 HRegion 集合中的一个 Column Family，建议把具有相近 IO 特性的 Column 存储在同一个 Column Family 中，以实现高效读取 。HStore 由一个 Memstore 及一系列 HFile 组成，Memstore 存储于内存当中，而 HFiles 则是写入到 HDFS 中的持久性文件。用户写入的数据首先会放入 MemStore，当 MemStore大小到达预设值（可通过 hbase.hregion.memstore.flush.size 进行配置）后就会 Flush 成一个StoreFile（即 HFile）文件。

2.2.3 MemStore

MemStore 是一个缓存 (In Memory Sorted Buffer)，当所有数据完成 WAL 日志写后，就会写入MemStore 中，由 MemStore 根据一定的算法将数据 Flush 到地层 HDFS 文件中(HFile)，每个 HRegion 中的每个 Column Family 有一个自己的 MemStore。当用户从 HBase 中读取数据时，系统将尝试从 MemStore 中读取数据，如果没找到相应数据才会尝试从 HFile 中读取。当服务器宕机时，MemStore 中的数据有可能会丢失，此时 HBase 就会使用 WAL 中的记录对 MemStroe 中的数据进行恢复。

2.2.4 HFile

HFile 是最终保存 HBase 数据行的文件，一个 HFile 文件属于一张表中的某个列簇，当中的数据是按 RowKey、Column Family、Column 升序排序，对相同的Cell(即这三个值都一样)，则按timestamp倒序排列。HFile 的具体格式如下图：

HFile 中每条键值存储的开发都包括2个固定长度的数字，分别表示键和值的长度，目的是让客户端可根据字节的偏移访问值域中的数据。根据上图可以看到 KeyValue 类中getKey和getRow方法的区别， getKey() 方法返回的是整个键（图中绿色部分），而 getRow() 方法返回的只是行键 RowKey（图中第4格）。
HFile 文件是根据表的列簇进行区分的，在执行持久化时，记录是有序的。但当 HFile 的文件内容增长到一定阈值后就会触发合并操作，多个 HFiles 就会合并成一个更大的 HFile，由于这几个 HFile 有可能在不同的时间段产生，为保证合并后数据依然是有序排列，HFile 会通过小量压缩或全量压缩进行合并，对 HFile 文件记录进行重新排序。由于全量压缩是一个耗费资源的操作，因此应该保证在资源充足的情况下进行（由于数据压缩问题已超出本文的界限，在以后的章节将会详细介绍）。
当单个 HFile 大小超过一定阈值后，会触发 Split 拆分操作，用户可通过配置 hbase.regionserver.region.split.policy 选择拆分的策略，拆分策略由 RegionSplitPolicy 类进行处理，目前系统已支持 IncreasingToUpperBoundRegionSplitPolicy、ConstantSizeRegionSplitPolicy、DelimitedKeyPrefixRegionSplitPolicy、KeyPrefixRegionSplitPolicy 等多种拆分策略。默认情况下 HRegion 将被拆分成 2 个 HRegion，父 HRegion 会下线，新分出的 2 个子 HRegion 会被 HMaster 分配到相应的 HRegionServer。

2.2.5 WAL

WAL（Write-Ahead-Log，又名 HLog）是 HRegionServer 中的日志记录的工具，当系统发生故障时，可以通过 WAL 恢复数据。在每次用户操作将数据写入MemStore的时候，也会写一份数据到 WAL 当中，WAL 当中包含了部分还没有写入 HFile 的文件。WAL 文件会定期滚动刷新，并删除旧的文件（已持久化到 HFile中的数据）。当 HMaster 通过 Zookeeper 感知到某个HRegionServer意外终止时，HMaster首先会处理遗留的 WAL 文件，将其中不同 HRegion 的 WAL 数据进行拆分，分别放到相应 WAL 的目录下，然后再将失效的 HRegion 重新分配，领取到这些 HRegion 的 HRegionServer 在加载 HRegion的过程中，会发现有历史 WAL 需要处理，因此会把 WAL 中的数据加载到 MemStore 中，然后 Flush 到HFiles，完成数据恢复。
用户可以通过禁用 WAL 方式提高HBase 的性能，然而这将有导致数据丢失的风险，用户应该谨慎处理。一旦禁用了 WAL，系统应该在接收到 HRegionServer 宕机的消息后重新启动写入程序，然而这有可能导致数据重复输入。

返回目录

三、简述 RowKey 设计原理

由于HBase属于分布式系统，数据会根据 RowKey 进行分块存储，只要合理设计好 RowKey 让数据均匀分散多台 HRegionServer 管理，时常被同时查找的数据被存储到同一个HRegion之内，这样就能最大限度地提高系统的性能（在第四节介绍到分页查询方法中，如果能巧妙地设计 RowKey 把每页的数据存储在一个HRegion之内，将有效地提升性能）。反之，如若 RowKey 设计不合理，让大量的数据存储在同一个 HRegionServer 而其他 HRegionServer 长期处理闲置状态，就会减低系统性能，还有可能让 HRegionSever 资源耗尽会发生错误。在拜读过 Sameer Wadkar 等大师所著的《Pro Apache Hadoop》和Nick Dimiduk 所著的 《HBase In Action》等文献后，本人对关于RowKey设计原则归纳成了下面几点：

  • 注意 RowKey 的长度，RowKey越长系统I/O开销越大，在上千万条数据的系统当中RowKey设置超过100个字节，光RowKey就会消耗近1G 的流量
  • 可将 Rowkey 的前位作为散列字段，由程序循环生成，扩展位放时间、表格、属性、时间戳等字段，这样可提高数据均衡分布在每个HRegionServer 实现负载均衡的几率。
  • 利用组合式行键方式，以主机名，事件，时间戳作为行键，根据组件的访问特性进行排序分组。

以随机的散列作为前缀这点很好理解，例如有下面的Order表

 OrderId　　  Goods  Price
 20180802001  iPhone X　　 8000
 20180802002  LYNK&CO Tools 5000
20180823003 AUSU N75S　　 8600
20180713004 Nikon D7500  7300

若只以OrderId作为RowKey，以递增方式进行存储，数据很有可能只存储于同一个HRegionServer，此时可以用Hash函数随机生成散列，加上必要的属性（可以是辨别符、时间戳等唯一属性），生成类似于154432_180802001、879531_180802002、544688_180823003、687851_180713004等数据。此数据的分配会更均衡，但查找时会更耗资源。
我们也可以利用主机名，事件，时间戳组合键的方式定制行键，例如系统配置了4个HRegionServer，分别用A、B、C、D代表，此时可以将 RowKey 配置为类似于 A-84548454-C、B-3223265-C、C-656565-C、D-333256-C，这里只是举个简单的例子，当然实际操作上组合方式有多种。此方法好处在于用户更容易地控制数据的分布，但会增加管理的繁琐度，如果服务器有变动时，数据存储可能需要重新整理。
其实RowKey设计本来就是一个复杂的问题，在这里介绍的只是冰山一角，希望对大家的RowKey设计有所启发。

返回目录

四、HBase 的 Java API 开发实例

4.1 HBase 的基础操作类

由于各版 HBase 的 Java API 会有不同，下面以比较稳定的  hbase-client 1.2.6 为例子介绍一下 HBase 的具体操作

4.1.1 org.apache.hadoop.hbase.HBaseConfiguration 类

继承了 org.apache.hadoop.conf.Configuration 类，主要用于配置系统运行环境，管理资源池

函数 描述
static Configuration create（） 获取当前运行环境下的 Configuration 配置对象
void addResource(Path file) 通过给定的路径所指的文件来添加资源
void clear() 清空所有已设置的属性
String get(String name) 获取属性名对应的值
String getBoolean(String name, boolean defaultValue) 获取为boolean类型的属性值，如果其属性值类型部位boolean,则返回默认属性值
void set(String name, String value) 通过属性名来设置值
void setBoolean(String name, boolean value) 设置boolean类型的属性值

 下面是HBaseConfiguration常用的方式

 1 public class HBaseUtils {
 2     public static Configuration config;
 3     public static Connection connection;
 4     
 5     static{
 6         config=HBaseConfiguration.create();
 7         
 8         try {
 9             connection=ConnectionFactory.createConnection(config);
10             
11         } catch (IOException e) {
12             // TODO 自动生成的 catch 块
13             e.printStackTrace();
14         }
15     }
16     ......
17 }

4.1.2 org.apache.hadoop.hbase.client.Connection 接口

与 SQL 的 Connection 连接相似，用户管理 HBase 客户端与服务端的连接，在操作完成后可通过 connetion.close （）及时释放资源。
通过 Connection 类的 Admin getAdmin（）方法可获取 Admin 管理类。
通过 ConnetionFactory 类的 static Connection createConnection (config)  静态方法可获取当前连接。

4.1.3 org.apache.hadoop.hbase.client.HBaseAdmin 类

用于管理HBase数据库的表信息，它提供的方法包括：创建表，删除表，列出表，使表有效或无效，以及添加或删除表列簇成员等。

方法 说明
void addColumn(String tableName, HColumnDescriptor column) 向一个已经存在的表添加列
static void checkHBaseAvailable(HBaseConfiguration conf) 静态函数，查看HBase是否处于运行状态
void createTable(HTableDescriptor desc) 创建一个表，同步操作
void deleteTable(String tableName) 删除一个已经存在的表
void enableTable(String tableName) 使表处于有效状态
void disableTable(String  tableName) 使表处于无效状态
HTableDescription listTables() 列出所有用户表
void modifyTable(byte[] tableName, HTableDescriptor htd) 修改表的模式，是异步的操作，可能需要花费一定的时间
boolean tableExists(String tableName) 检查表是否存在

 下面例子可用于判断表格是否存在

 1 public class HBaseUtils {
 2     public static Configuration config;
 3     public static Connection connection;
 4     
 5     static{
 6         config=HBaseConfiguration.create();
 7         
 8         try {
 9             connection=ConnectionFactory.createConnection(config);
10         } catch (IOException e) {
11             // TODO 自动生成的 catch 块
12             e.printStackTrace();
13         }
14     }
15 
16     public static void checkTable(String tableName)
17             throws Exception {
18         HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
19  
20         if (admin.tableExists(tableName)) {
21             System.out.println(tableName + " exists!");
22         }
23     }
24 }

4.1.4 org.apache.hadoop.hbase.HTableDescriptor 类

用于操作表名及其对应表的列簇

方法 说明
HTableDescriptor addFamily(HColumnDescriptor) 添加一个列簇
HColumnDescriptor removeFamily(byte[] column) 移除一个列簇
String getNameAsString() 获取表的名字
String getValue(String) 获取属性的值
HTableDescriptor setValue(String key, String value) 设置属性的值

4.1.5 org.apache.hadoop.hbase.HColumnDescriptor 类

维护列簇的信息，例如版本号，压缩设置等，通常在创建表或者为表添加列簇的时候使用。
列簇被创建后不能直接修改，只能通过删除然后重新创建的方式。 列簇被删除的时候，列簇里面的数据也会同时被删除。

方法 说明
String getNameAsString() 获取列簇的名字
String getValue(String) 获取对应的属性的值
HColumnDescriptor setValue(String key, String value) 设置对应属性的值

一个表通常可以包含1~5个列簇，下面例子可用于新建表，如需要包含多个列簇，可以在columnFamily参数中用 “ , ” 输入

 1     //新建表
 2     public static boolean createTable(String tableName, String columnFamily)
 3             throws Exception {
 4         HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
 5  
 6         if (admin.tableExists(tableName)) {
 7             System.out.println(tableName + " exists!");
 8             return false;
 9         } else {
10             //建立列簇
11             String[] columnFamilyArray;
12             if(columnFamily.contains(","))
13                 columnFamilyArray = columnFamily.split(",");
14             else{
15                 columnFamilyArray=new String[1];
16                 columnFamilyArray[0]=columnFamily;
17             }
18             HColumnDescriptor[] hColumnDescriptor = new HColumnDescriptor[columnFamilyArray.length];
19             for (int i = 0; i < hColumnDescriptor.length; i++) {
20                 hColumnDescriptor[i] = new HColumnDescriptor(columnFamilyArray[i]);
21             }
22             //建立表对象
23             HTableDescriptor familyDesc = new HTableDescriptor(TableName.valueOf(tableName));
24             for (HColumnDescriptor columnDescriptor : hColumnDescriptor) {
25                 familyDesc.addFamily(columnDescriptor);
26             }
27             HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(tableName), familyDesc);
28             //新建表
29             admin.createTable(tableDesc);
30             admin.close();
31             return true;
32         }
33     }

4.1.6 org.apache.hadoop.hbase.client.Table 接口

由于 HTable 是非线性安全类，已经被系统丢弃，在 hbase-client v1.0 版本后在数据更新删除时应该使用线性安全的 Table 接口进行操作

方法 说明
boolean checkAdnPut(byte[] row, byte[] family, byte[] qualifier, byte[] value, Put put 自动的检查row/family/qualifier是否与给定的值匹配
void close() 释放所有的资源或挂起内部缓冲区中的更新，请谨记在数据更新删除后使用以释放资源
boolean exists(Get get) 检查Get实例所指定的值是否存在于Table中
Result get(Get get) 获取指定行的某些单元格所对应的值
Result[] get(List<Get> list) 获取多行的单元格所对应的值
ResultScanner getScanner(Scan scan) 获取当前给定列簇的scanner实例
HTableDescriptor getTableDescriptor() 获取当前表的HTableDescriptor实例
void delete(Delete delete) 删除数据
void delete(List<Delete> list) 删除多行数据
void put(List<Put> list) 向表中添加多个值
void put(Put put) 向表中添加值

此接口是HBase最常用的对表格操作的接口，具体的使用方法将在下一节再详细介绍

4.1.7 org.apache.hadoop.hbase.client.Put 类

主要用于对单个行执行添加操作，在 hbase-client v1.0 版本后 add 方法已经被 addColumn 方法所代替

方法 说明
Put addColumn(byte[] family, byte[] qualifier, byte[] value) 将指定的列和对应的值添加到Put实例中
Put addColumn(byte[] family, byte[] qualifier, long ts, byte[] value) 将指定的列和对应的值及时间戳添加到Put实例中
List<Cell> get(byte[] family,byte[] qualifier) 获取指定列簇和对应值的列

4.1.8 org.apache.hadoop.hbase.client.Get 类

用于获取单行数据的相关信息

方法 说明
Get addColumn(byte[] family, byte[] qualifier) 获取指定列簇和列修饰符对应的Get对象
Get addFamily(byte[] family) 获取指定列簇对应列的Get对象
Get setTimeRange(long timeStamp) 获取指定时间戳的Get对象
Get setFilter(Filter filter) 执行Get操作时设置服务器端的过滤器

4.1.9 org.apache.hadoop.hbase.client.Delete类

用于获取单行数据的相关信息

方法 说明
Delete addColumn(byte[] family, byte[] qualifier) 获取指定列簇和列修饰符的Delete对象（只包含当前version）
Delete addColumn(byte[] family, byte[] qualifier,long timestamp) 获取指定列簇、列修饰符和时间戳的Delete对象（只包含当前version）
Delete addColumns(byte[] family, byte[] qualifier) 获取指定列簇和列修饰符（包含所有version）
Delete addColumns(byte[] family, byte[] qualifier,long timestamp) 获取指定列簇、列修饰符和时间戳的Delete对象（包含所有version）
Delete addFamily(byte[] family) 获取指定的列簇的Delete对象
Delete setTimeRange(long timeStamp) 获取指定时间戳的Delete对象

4.1.10 org.apache.hadoop.hbase.client.Result 类

用于封装数据查找后的单行查询结果，用 Map 结构保存符合列的对应属性。旧版本一直使用KeyValue来封装列的属性，但自从v1.0版本后，系统都会使用 Cell 对象来封装列的属性，多个旧方法已经被弃用，各位在使用时可以注意一下

方法 说明
byte[] getRow（） 获取当前行键值
byte[] getValue(byte[] family, byte[] qualifier) 获取指定列簇和列修饰符的最新数据值
Cell getColumnLastestCell(byte[] family, byte[] qualifier) 获取指定列簇、列修饰符最新版本的列
List<Cell> getColumnCells(byte[] family, byte[] qualifier) 获取指定列簇、列修饰符多个版本的列
boolean containsColumn(byte[] family, byte[] qualifier) 判断是否存在指定列簇、列修饰符的列
Cell[] listCells（）　　 获取多个版本所有列
NavigableMap<byte[], byte[]> getFamilyMap(byte[] family) 获取批定列簇的列
NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> getMap() 获取所有列簇所有版本的列

4.1.11 org.apache.hadoop.hbase.client.Scan类

当用户想在查询时一次返回多选结果，可以通过Scan进行条件查询，通过Scan可以设定多种过滤器和起始行，从而根据用户的需要对整张表进行扫描。然而在默认情况下，Scan对象是用迭代形式进行查询一次只返回一行，为了提高内存的使用率可以通过 Scan.setCaching (int caching）方法把返回行数进行调整，从而提高系统性能。

方法 说明
Scan addFamily（byte[] family） 获取指定列簇的Scan
Scan addColumn(byte[] family, byte[] qualifier) 获取指定列簇和列修饰符的Scan
Scan setCaching(int caching) 设置缓存数量
Scan setStartRow（byte[] startRow) 设置开始行rowkey，如果不设置将由第一行开始查找
Scan setStopRow（byte[] stopRow) 设置结束行rowkey
Scan setTimeStamp(long stamp) 获取指定时间戳的Scan对象
Scan setMaxResultSize(long maxResultSize) 设置最大返回数量
Scan setFilter（Filter filter) 设置查询条件

4.1.12 org.apache.hadoop.hbase.client.ResultScanner 接口

由于Scan 查找会返回多行数据，为了实现逐行查询功能，ResultScanner类出现，它把扫描到每一行数据封装成一个Result实例，并将所有的Result实例放入一个迭代器中。
ResultScanner 只有几个简单的方法，在有条件的情况下，使用Result[] next（int paramInt）一次获取多条数据，有利用提升系统的性能。

方法 说明
Result next（） 返回Result，转到下一行
Result[] next (int paramInt) 返回多行数据的Result数组
void close（） 关闭连接释放资源

4.2 通用类 HBaseUtils

下面本人参考了一些基础文献对HBase常用的CURD操作进行了归纳，整理出一个通用类HBaseUtils，希望对大家日常开发有所帮助。
由于HBase是分布式系统，其性能与数据的存储方式有莫大的关联，所以在开发时应该与第三节RowKey的设计原理相结合进行调整，要不然有可能影响系统性能。

  1 public class HBaseUtils {
  2     public static Configuration config;
  3     public static Connection connection;
  4     
  5     static{
  6         config=HBaseConfiguration.create();
  7         
  8         try {
  9             connection=ConnectionFactory.createConnection(config);
 10         } catch (IOException e) {
 11             // TODO 自动生成的 catch 块
 12             e.printStackTrace();
 13         }
 14     }
 15 
 16     //新建表
 17     public static boolean createTable(String tableName, String columnFamily)
 18             throws Exception {
 19         HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
 20  
 21         if (admin.tableExists(tableName)) {
 22             System.out.println(tableName + " exists!");
 23             return false;
 24         } else {
 25             //建立列簇
 26             String[] columnFamilyArray;
 27             if(columnFamily.contains(","))
 28                 columnFamilyArray = columnFamily.split(",");
 29             else{
 30                 columnFamilyArray=new String[1];
 31                 columnFamilyArray[0]=columnFamily;
 32             }
 33             HColumnDescriptor[] hColumnDescriptor = new HColumnDescriptor[columnFamilyArray.length];
 34             for (int i = 0; i < hColumnDescriptor.length; i++) {
 35                 hColumnDescriptor[i] = new HColumnDescriptor(columnFamilyArray[i]);
 36             }
 37             //建立表对象
 38             HTableDescriptor familyDesc = new HTableDescriptor(TableName.valueOf(tableName));
 39             for (HColumnDescriptor columnDescriptor : hColumnDescriptor) {
 40                 familyDesc.addFamily(columnDescriptor);
 41             }
 42             HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(tableName), familyDesc);
 43             //新建表
 44             admin.createTable(tableDesc);
 45             admin.close();
 46             return true;
 47         }
 48     }
 49  
 50     //插入数据
 51     public static boolean put(String tablename, String row, String columnFamily,
 52                               String qualifier, String data) throws Exception {
 53         Table table = connection.getTable(TableName.valueOf(tablename));
 54         Put put = new Put(Bytes.toBytes(row));
 55         put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier),
 56                 Bytes.toBytes(data));
 57         table.put(put);
 58         System.out.println("put \'" + row + "\', \'" + columnFamily + ":" + qualifier
 59                 + "\', \'" + data + "\'");
 60         table.close();
 61         return true;
 62     }
 63     
 64     //插入多列数据
 65     public static boolean put(String tablename, String row, String columnFamily,
 66                               String[] qualifierList, String[] dataList) throws Exception {
 67         Table table = connection.getTable(TableName.valueOf(tablename));
 68         Put put = new Put(Bytes.toBytes(row));
 69         for(int n=0;n<qualifierList.length;n++){
 70             put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifierList[n]),
 71                   Bytes.toBytes(dataList[n]));
 72         }
 73         table.put(put);
 74         table.close();
 75         return true;
 76     }
 77  
 78     //查看某行
 79     public static Map<String, Object> getRow(String tablename, String row) throws Exception {
 80         Table table = connection.getTable(TableName.valueOf(tablename));
 81         Get get = new Get(Bytes.toBytes(row));
 82         Result result = table.get(get);
 83         table.close();
 84         return resultToMap(result);
 85     }
 86  
 87     //查看全表
 88     public static List<Map<String, Object>> getTable(String tablename) throws Exception {
 89         Table table = connection.getTable(TableName.valueOf(tablename));
 90         Scan s = new Scan();
 91         ResultScanner rs = table.getScanner(s);
 92  
 93         List<Map<String, Object>> resList = new ArrayList<Map<String, Object>>();
 94         for (Result r : rs) {
 95             Map<String, Object> tempmap = resultToMap(r);
 96             resList.add(tempmap);
 97         }
 98         table.close();
 99         return resList;
100     }
101     
102     //单个qualifier的值等于data
103     public static List<Map<String, Object>> queryEqual(String tablename, String columnFamily, String qualifier, String data) throws Exception {
104         //某列等于data的数据
105         Filter filter = new SingleColumnValueFilter(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier),
106                 CompareOp.EQUAL, Bytes.toBytes(data));
107         FilterList filterList = new FilterList();
108         filterList.addFilter(filter);
109         return query(tablename, filterList);
110     }
111  
112     //查询qualifier值在mindata和maxdata之间的数据
113     public static List<Map<String, Object>> queryBetween(String tablename, String columnFamily, String qualifier, String mindata, String maxdata) throws Exception {
114         SingleColumnValueFilter filter1 = new SingleColumnValueFilter(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier),
115                 CompareOp.LESS_OR_EQUAL, Bytes.toBytes(maxdata));
116         SingleColumnValueFilter filter2 = new SingleColumnValueFilter(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier),
117                 CompareOp.GREATER_OR_EQUAL, Bytes.toBytes(mindata));
118         FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
119         //不存在此值时跳过
120         filter1.setFilterIfMissing(true);
121         filter2.setFilterIfMissing(true);
122         //加入过虑器
123         filterList.addFilter(filter1);
124         filterList.addFilter(filter2);
125         return query(tablename, filterList);
126     }
127  
128     //查询列
129     public static List<Map<String, Object>> queryColumn(String tablename, String prefix) throws Exception {
130         Filter filter = new ColumnPrefixFilter(Bytes.toBytes(prefix));
131         FilterList filterList = new FilterList();
132         filterList.addFilter(filter);
133         return query(tablename, filterList);
134     }
135  
136     //分页查询
137     public static List<Map<String, Object>> queryRowPage(String tablename, String columnFamily, String qualifier,String value,String count,String startrowkey,String stoprowkey) throws Exception {
138         FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
139         //设置每页面数量pageCount
140         Filter filter1 = new PageFilter(Integer.parseInt(count));
141         Filter filter2 = new SingleColumnValueFilter(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier),
142                 CompareOp.EQUAL, Bytes.toBytes(value));
143         //设置查询条件
144         filterList.addFilter(filter1);
145         filterList.addFilter(filter2);
146  
147         Table table = connection.getTable(TableName.valueOf(tablename));
148         Scan s=new Scan();
149         //设置key的开始值和结束值
150         if(startrowkey!=null)
151            s.setStartRow(Bytes.toBytes(startrowkey));
152         if(stoprowkey!=null)
153            s.setStopRow(Bytes.toBytes(stoprowkey));
154         //加入查询条件
155         s.setFilter(filterList);
156         ResultScanner rs = table.getScanner(s);
157  
158         List<Map<String, Object>> resList = new ArrayList<Map<String, Object>>();
159         for (Result r : rs) {
160             Map<String, Object> tempmap = resultToMap(r);
161             resList.add(tempmap);
162         }
163 
164         table.close();
165         return resList;
166     }
167     
168     //数据转换为Map
169     private static Map<String, Object> resultToMap(Result result) {
170         Map<String, Object> resMap = new HashMap<String, Object>();
171         List<Cell> listCell = result.listCells();
172         Map<String, Object> tempMap = new HashMap<String, Object>();
173         String rowname = "";
174         List<String> familynamelist = new ArrayList<String>();
175         for (Cell cell : listCell) {
176             byte[] rowArray = cell.getRowArray();
177             byte[] familyArray = cell.getFamilyArray();
178             byte[] qualifierArray = cell.getQualifierArray();
179             byte[] valueArray = cell.getValueArray();
180             int rowoffset = cell.getRowOffset();
181             int familyoffset = cell.getFamilyOffset();
182             int qualifieroffset = cell.getQualifierOffset();
183             int valueoffset = cell.getValueOffset();
184             int rowlength = cell.getRowLength();
185             int familylength = cell.getFamilyLength();
186             int qualifierlength = cell.getQualifierLength();
187             int valuelength = cell.getValueLength();
188  
189             byte[] temprowarray = new byte[rowlength];
190             System.arraycopy(rowArray, rowoffset, temprowarray, 0, rowlength);
191             String temprow= Bytes.toString(temprowarray);
192 
193             byte[] tempqulifierarray = new byte[qualifierlength];
194             System.arraycopy(qualifierArray, qualifieroffset, tempqulifierarray, 0, qualifierlength);
195             String tempqulifier= Bytes.toString(tempqulifierarray);
196  
197             byte[] tempfamilyarray = new byte[familylength];
198             System.arraycopy(familyArray, familyoffset, tempfamilyarray, 0, familylength);
199             String tempfamily= Bytes.toString(tempfamilyarray);
200  
201             byte[] tempvaluearray = new byte[valuelength];
202             System.arraycopy(valueArray, valueoffset, tempvaluearray, 0, valuelength);
203             String tempvalue= Bytes.toString(tempvaluearray);
204  
205             tempMap.put(tempfamily + ":" + tempqulifier, tempvalue);
206 
207             rowname = temprow;
208             String familyname = tempfamily;
209             if (familynamelist.indexOf(familyname) < 0) {
210                 familynamelist.add(familyname);
211             }
212         }
213         resMap.put("rowname", rowname);
214         for (String familyname : familynamelist) {
215             HashMap<String,Object> tempFilterMap = new HashMap<String,Object>();
216             for (String key : tempMap.keySet()) {
217                 String[] keyArray = key.split(":");
218                 if(keyArray[0].equals(familyname)){
219                     tempFilterMap.put(keyArray[1],tempMap.get(key));
220                 }
221             }
222             resMap.put(familyname, tempFilterMap);
223         }
224 
225         return resMap;
226     }
227  
228     //公共query查找方法
229     private static List<Map<String, Object>> query(String tablename, FilterList filterList) throws Exception {
230         Table table = connection.getTable(TableName.valueOf(tablename));
231         Scan s = new Scan();
232         s.setFilter(filterList);
233         ResultScanner rs = table.getScanner(s);
234  
235         List<Map<String, Object>> resList = new ArrayList<Map<String, Object>>();
236         for (Result r : rs) {
237             Map<String, Object> tempmap = resultToMap(r);
238             resList.add(tempmap);
239         }
240         table.close();
241         return resList;
242     }
243 
244  
245     //删除表
246     public static boolean delete(String tableName) throws IOException {
247         HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
248         if (admin.tableExists(tableName)) {
249             try {
250                 admin.disableTable(tableName);
251                 admin.deleteTable(tableName);
252                 admin.close();
253             } catch (Exception e) {
254                 e.printStackTrace();
255                 return false;
256             }finally{
257                 admin.close();
258             }
259         }
260         return true;
261     }
262  
263     //删除ColumnFamily
264     public static boolean deleteColumnFamily(String tableName,String columnFamilyName) throws IOException {
265         HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
266         if (admin.tableExists(tableName)) {
267             try {
268                 admin.deleteColumn(tableName,columnFamilyName);
269             } catch (Exception e) {
270                 e.printStackTrace();
271                 return false;
272             }finally{
273                 admin.close();
274             }
275         }
276         return true;
277     }
278     
279     //删除row
280     public static boolean deleteRow(String tableName,String rowName) throws IOException {
281         HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
282         Table table = connection.getTable(TableName.valueOf(tableName));
283         if (admin.tableExists(tableName)) {
284             try {
285                 Delete delete = new Delete(rowName.getBytes());
286                 table.delete(delete);
287             } catch (Exception e) {
288                 e.printStackTrace();
289                 return false;
290             }finally{
291                 table.close();
292             }
293         }
294         return true;
295     }
296  
297     //删除qualifier
298     public static boolean deleteQualifier(String tableName,String rowName,String columnFamilyName,String qualifierName) throws IOException {
299         HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
300         Table table = connection.getTable(TableName.valueOf(tableName));
301         if (admin.tableExists(tableName)) {
302             try {
303                 Delete delete = new Delete(rowName.getBytes());
304                 delete.addColumns(columnFamilyName.getBytes(),qualifierName.getBytes());
305                 table.delete(delete);
306             } catch (Exception e) {
307                 e.printStackTrace();
308                 return false;
309             }finally{
310                 table.close();
311             }
312         }
313         return true;
314     }
315     
316 }

返回目录

五、HBase与MapReduce的相互调用

5.1 MapReduce 与 HBase 的关系

有朋友常说现在流程NoSQL架构 ,系统常用的是 Spark、HBase、Redis、Cassandra 等技术，MapReducer 已经没落。然而本人觉得 MapReduce 与 HBase 等NoSQL技术并非处于对立发展的关系，反而是相辅相成，互补不足的技术。由于受到数据块、IO、网络资源等限制，MapReduce 的先天架构决定其主要应用于大型文件的存取管理。然而对大量小型文件的管理方面，MapReduce 却是无能为力。因此 HBase 等技术才会诞生并迅速发展，随着 Hadoop 2.0 的发展，HBase 以 HDFS 为基础，补尝了 MapReduce 在海量文件管理方面的不足。
在很多大型的*机关，金融管理，图书管理，交通航运等平台上，往往会以 MR 作为大型的持久性资源库，以 HBase 进行数据提取作为某个区域或阶段的研究对象，加以分析挖掘，最后形成汇总。有见及此，Hadoop 为 MR与HBase开发出一系列的数据转换工具，方便开发人员利用。 

5.2 常用类简介

5.2.1 TableMapReduceUtil 绑定表关系

TableMapReduceUtil为编辑人员准备的几个常用的方法，可快速地绑定Mapper/Reducer与HBase中Table的关系

方法 说明
static void setScannerCaching(Job job, int batchSize) 设置Scan缓存大小，默认值为1，应按运行环境设置不适宜过大
static void initTableReducerJob(String table,Class<? extends TableReducer> reducer, Job job) 绑定从Reducer输出数据所要存入的Table
static void initTableMapperJob(String table, Scan scan,Class<? extends TableMapper> mapper, Class<?> outputKeyClass,Class<?> outputValueClass, Job job) 绑定要从Table获取到Scan数据所送往的Mapper
static void initTableMapperJob(List<Scan> scans,Class<? extends TableMapper> mapper,Class<?> outputKeyClass,Class<?> outputValueClass, Job job) 绑定要从多个Table中获取到的List<Scan>所送往的Mapper

5.2.2 设置数据格式

在设置Job的运行条件时可使用以下方法按照需要把数据输入或输入设置为Table格式

Job.setInputFormatClass(TableOutputFormat.class);
Job.setOutputFormatClass(TableOutputFormat.class);

5.2.3 TableMapper 与 TableReducer

为提供与 HBase 的数据对接，系统提供了TableMapper与TableReducer两个常用类

public abstract class TableMapper<KEYOUT, VALUEOUT>
       extends Mapper<ImmutableBytesWritable, Result, KEYOUT, VALUEOUT> {
}

TableMappler 是继承了 Mapper<ImmutableBytesWritable, Result, KEYOUT, VALUEOUT> 并以 ImmeutableBytesWritable对象（键值RowKey）为输入InputKey，以 Result 为输入 InputValue 的 Mapper，目的是把HBase中某个Table中的行数据输入到 TableMapper 进行处理

public abstract class TableReducer<KEYIN, VALUEIN, KEYOUT>
        extends Reducer<KEYIN, VALUEIN, KEYOUT, Mutation> {
}

TableReducer 则继承了 Reducer<KEYIN, VALUEIN, KEYOUT, Mutation>，它以 RowKey 作出输出键 OutputKey，以行数据 Mutation 作为输出值 OutputValue ，把 TableReducer 处理完的数据发送到预先绑定的 Table 中。 Mutation 虚拟类实现了 Row, CellScannable 等接口（Put类就是Mutation最常用的实现类），因此系统可以根据 RowKey 轻松地找到对应的行数据。

5.3 常用实例

5.3.1 利用TableReducer从MapReduce中提取数据发送到HBase中的某个Table

假设某机关单位用MR存储了全国人员的申请审核文件档案，各个省镇部门需要按照自己权限获取相应的资料保存到各自的HBase中

此时，可以在Mapper 中按时地区参数 Area 进行筛选，把符合条件的资料发送到 TableReducer, 通过 TaskInputOutputContext.write(ImmutableBytesWritable arg0, Mutation arg1) 方法，把Reducer 中的数据保存到 HBase 的 Table 中去。其中 ImmutableBytesWritable 为行键 RowKey，Mutation 为行数据。 

  1 public class MrToHBaseExample extends Configured implements Tool{
  2     
  3     public static class MyMapper extends Mapper<LongWritable,Text,NullWritable,ExamineWritable>{
  4         private static String Area="nothing";
  5         
  6         @Override
  7         public void setup(Context context){
  8             String param=context.getConfiguration().getStrings("Area")[0];
  9             if(!param.isEmpty())
 10                 Area=param;
 11         }
 12         
 13         public void map(LongWritable longwritable,Text text,Context context)
 14             throws IOException,InterruptedException{
 15             String[] data=text.toString().split(",");
 16             ExamineWritable examine=new ExamineWritable();
 17             examine.Id=new Text(data[0]);
 18             examine.Name=new Text(data[1]);
 19             examine.Age=new IntWritable(new Integer(data[2]));
 20             examine.Gender=new Text(data[3]);
 21             examine.Area=new Text(data[4]);
 22             examine.Approved=new IntWritable(new Integer(data[5]));
 23             if(Area.equals(data[4])){
 24                 context.write(NullWritable.get(), examine);
 25             }
 26         }      
 27     }
 28     
 29     public static class MyReducer extends TableReducer<NullWritable,ExamineWritable,ImmutableBytesWritable>{        
 30         public void reduce(NullWritable key,Iterable<ExamineWritable> values,Context context)
 31             throws IOException,InterruptedException{
 32             
 33             for(ExamineWritable value : values){
 34                 Put put=new Put(Bytes.toBytes(value.Id.toString()));
 35                 put.addColumn(Bytes.toBytes("family1"), Bytes.toBytes("id"), Bytes.toBytes(value.Id.toString()));
 36                 put.addColumn(Bytes.toBytes("family1"), Bytes.toBytes("name"), Bytes.toBytes(value.Name.toString()));
 37                 put.addColumn(Bytes.toBytes("family1"), Bytes.toBytes("age"), Bytes.toBytes(value.Age.toString()));
 38                 put.addColumn(Bytes.toBytes("family1"), Bytes.toBytes("gender"), Bytes.toBytes(value.Gender.toString()));
 39                 put.addColumn(Bytes.toBytes("family1"), Bytes.toBytes("area"), Bytes.toBytes(value.Area.toString()));
 40                 put.addColumn(Bytes.toBytes("family1"), Bytes.toBytes("approved"), Bytes.toBytes(value.Approved.toString()));
 41                 context.write(new ImmutableBytesWritable(Bytes.toBytes(value.Id.toString())), put);
 42             }
 43         }
 44     }
 45 
 46     public int run(String[] arg0) throws Exception {
 47         // TODO 自动生成的方法存根
 48          // TODO Auto-generated method stub
 49         //生成表Examine,列簇family1
 50         HBaseUtils.createTable("Examine", "family1");
 51         //建立任务Job
 52         Job job=Job.getInstance(getConf());
 53         job.setJarByClass(MrToHBaseExample.class);
 54         //初始化HBase,把MyReducer输入数据保存到Examine表
 55         TableMapReduceUtil.initTableReducerJob("Examine", MyReducer.class, job);
 56         //注册Key/Value类型为Text
 57         job.setOutputKeyClass(ImmutableBytesWritable.class);
 58         job.setOutputValueClass(Put.class);
 59         //若Map的转出Key/Value不相同是需要分别注册
 60         job.setMapOutputKeyClass(NullWritable.class);
 61         job.setMapOutputValueClass(ExamineWritable.class);
 62         //注册Mapper及Reducer处理类
 63         job.setMapperClass(MyMapper.class);
 64         job.setReducerClass(MyReducer.class);
 65         //输入输出数据格式化类型为TextInputFormat/TableOutputFormat
 66         job.setInputFormatClass(TextInputFormat.class);
 67         job.setOutputFormatClass(TableOutputFormat.class);
 68         //获取命令参数
 69         String[] args=new GenericOptionsParser(getConf(),arg0).getRemainingArgs();
 70         FileInputFormat.setInputPaths(job,new Path(args[0]));    
 71         boolean status=job.waitForCompletion(true);
 72         if(status)
 73             return 0;
 74         else
 75             return 1;
 76     }
 77     
 78     public static void main(String[] args) throws Exception{
 79         Configuration conf=new Configuration();
 80         ToolRunner.run(new MrToHBaseExample(), args);      
 81     } 
 82 }
 83 
 84 public class HBaseUtils {
 85     public static Configuration config;
 86     public static Connection connection;
 87     
 88     static{
 89         config=HBaseConfiguration.create();
 90         
 91         try {
 92             connection=ConnectionFactory.createConnection(config);
 93         } catch (IOException e) {
 94             // TODO 自动生成的 catch 块
 95             e.printStackTrace();
 96         }
 97     }
 98 
 99     //新建表
100     public static boolean createTable(String tableName, String columnFamily)
101             throws Exception {
102         HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
103  
104         if (admin.tableExists(tableName)) {
105             System.out.println(tableName + " exists!");
106             return false;
107         } else {
108             //建立列簇
109             String[] columnFamilyArray;
110             if(columnFamily.contains(","))
111                 columnFamilyArray = columnFamily.split(",");
112             else{
113                 columnFamilyArray=new String[1];
114                 columnFamilyArray[0]=columnFamily;
115             }
116             HColumnDescriptor[] hColumnDescriptor = new HColumnDescriptor[columnFamilyArray.length];
117             for (int i = 0; i < hColumnDescriptor.length; i++) {
118                 hColumnDescriptor[i] = new HColumnDescriptor(columnFamilyArray[i]);
119             }
120             //建立表对象
121             HTableDescriptor familyDesc = new HTableDescriptor(TableName.valueOf(tableName));
122             for (HColumnDescriptor columnDescriptor : hColumnDescriptor) {
123                 familyDesc.addFamily(columnDescriptor);
124             }
125             HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(tableName), familyDesc);
126             //新建表
127             admin.createTable(tableDesc);
128             admin.close();
129             return true;
130         }
131     }
132 }

只需要输入执行命令
hadoop jar 【Jar名称】 【Main类全名称】-D 【参数名=参数值】 【InputPath】 
即可得到以下运行结果（ 若要获取广东省范围内的文件，参数为 Area=GuangDong）

5.3.2 通过TableMapper把HBase中某个表的数据汇总到 MR

当系统需要把符合条件的数据从HBase汇总到MR时，可利用TableMapper工具类，先建立Scan对象，加入筛选条件，然后利用 static void initTableMapperJob(String table, Scan scan,Class<? extends TableMapper> mapper, Class<?> outputKeyClass,Class<?> outputValueClass, Job job) 方法绑定TableMapper，Scan，Job等关系，系统就会把 table 中符合筛选条件的数据发送Mapper 进行处理。
下面例子就是从HBase的Examine表中把Area等于GuangDong的数据发送到MR进行保存

 1 public class MrToHBaseExample extends Configured implements Tool{
 2     
 3     public static class MyMapper extends TableMapper<Text,ExamineWritable>{
 4         
 5         public void map(ImmutableBytesWritable writable,Result result,Context context)
 6             throws IOException,InterruptedException{            
 7             ExamineWritable examine=new ExamineWritable();
 8             for(Cell cell:result.listCells()){
 9                 String value=Bytes.toString(CellUtil.cloneValue(cell));
10                 //获取行数据
11                 if((Bytes.toString(CellUtil.cloneQualifier(cell))).equals("id"))
12                      examine.Id=new Text(value);
13                 if((Bytes.toString(CellUtil.cloneQualifier(cell))).equals("name"))
14                      examine.Name=new Text(value);
15                  if((Bytes.toString(CellUtil.cloneQualifier(cell))).equals("age"))
16                      examine.Age=new IntWritable(new Integer(value));
17                 if((Bytes.toString(CellUtil.cloneQualifier(cell))).equals("gender"))
18                     examine.Gender=new Text(value);
19                 if((Bytes.toString(CellUtil.cloneQualifier(cell))).equals("area"))
20                     examine.Area=new Text(value);
21                  if((Bytes.toString(CellUtil.cloneQualifier(cell))).equals("approved"))
22                      examine.Approved=new IntWritable(new Integer(value));
23             }
24 
25             context.write(new Text(Bytes.toString(writable.get())),examine);
26         }      
27     }
28     
29     public static class MyReducer extends Reducer<Text,ExamineWritable,NullWritable,Text>{        
30         public void reduce(Text key,Iterable<ExamineWritable> values,Context context)
31             throws IOException,InterruptedException{
32             String data=null;
33             for(ExamineWritable value : values){
34                 data=value.Id.toString()+","+value.Name.toString()+","
35                     +value.Age.toString()+","+value.Gender.toString()+","
36                     +value.Area.toString()+","+value.Approved.toString();
37                 context.write(NullWritable.get(),new Text(data));
38             }
39         }
40     }
41 
42     public int run(String[] arg0) throws Exception {
43         // TODO 自动生成的方法存根
44          // TODO Auto-generated method stub
45         //建立任务Job
46         Job job=Job.getInstance(getConf());
47         job.setJarByClass(MrToHBaseExample.class);
48         //初始化TableMapper,把Examine表中符合条件的数据发送到Mapper
49         Scan scan=new Scan();
50         scan.addFamily(Bytes.toBytes("family1"));
51         Filter filter=new SingleColumnValueFilter(Bytes.toBytes("family1"),Bytes.toBytes("Area"),CompareOp.EQUAL,Bytes.toBytes("GuangDong"));
52         scan.setFilter(filter);
53         TableMapReduceUtil.initTableMapperJob("Examine",scan,MyMapper.class,Text.class,ExamineWritable.class,job);
54         //注册Key/Value类型为Text
55         job.setOutputKeyClass(NullWritable.class);
56         job.setOutputValueClass(Text.class);
57         //若Map的转出Key/Value不相同是需要分别注册
58         job.setMapOutputKeyClass(Text.class);
59         job.setMapOutputValueClass(ExamineWritable.class);
60         //注册Mapper及Reducer处理类
61         job.setMapperClass(MyMapper.class);
62         job.setReducerClass(MyReducer.class);
63         //输入输出数据格式化类型为TextInputFormat/TableOutputFormat
64         job.setInputFormatClass(TableInputFormat.class);
65         job.setOutputFormatClass(TextOutputFormat.class);
66         //获取命令参数
67         String[] args=new GenericOptionsParser(getConf(),arg0).getRemainingArgs();
68         FileOutputFormat.setOutputPath(job,new Path(args[0]));    
69         boolean status=job.waitForCompletion(true);
70         if(status)
71             return 0;
72         else
73             return 1;
74     }
75     
76     public static void main(String[] args) throws Exception{
77         Configuration conf=new Configuration();
78         ToolRunner.run(new MrToHBaseExample(), args);      
79     } 
80 }

运行结果

5.3.2 通过TableMapper把HBase中多个表的数据汇总到 MR

若输入的数据源来源于HBase的不同表格，TableMapReduceUtil 类还提供static void initTableMapperJob(List<Scan> scans,Class<? extends TableMapper> mapper,Class<?> outputKeyClass,Class<?> outputValueClass, Job job) 方法来实现此功能。用户可在通过List<Scan>参数绑定不同的数据源，利用 Scan.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, byte[] tableName) 方法绑定Scan所要查找的表名。然后通过 Job.setInputFormatClass(MultiTableInputFormat.class) 设置为多数据源采集模式。若要在TableMapper中区分不同的数据源表格，还可以在上下文中通过TableSplit.getTableName()方法获取输入的表格名称。

假如在HBase中Examine表格保存了审核结果表，当中包含了申请编号id，申请人姓名 name，年龄 age，性别 gender，所在地 area，审核结果 approved等信息

Assess 表格保存了审核流程记录表,当中包含申请编号id，申请时间 filingDate，审核时间approvedDate，审核人 assessor，审核商品 goods等信息

系统会通过TableMapReduceUtil 类所提供的 static void initTableMapperJob(List<Scan> scans,Class<? extends TableMapper> mapper,Class<?> outputKeyClass,Class<?> outputValueClass, Job job) 方法绑定两个不同的 Table。重写 Mapper 的 setup() 方法，通过上下文context 获取TableSplit对象，然后利用TableSplit.getTableName()分辨所输入数据表的来源。然后把ExamineDetailKey 作为 TableMapper 的 OutputKey, 以 ExamineDetailValue作为TableMapper的OutputValue，通过自定义的ExamineDetailSortComparator，ExamineDetailGroupComparator 对数据进行排序与分组，把 Id 相同的一组数据分派到同一个reduce方法进行处理，最后把同一组数据的审核结果与流程记录保存到MR中。
由于本文目的主要是讲述 HBase 对表数据的处理方式，在此对排序SortComparator/分组GroupComparator不作细说，如果对 MR 的数据排序组合操作有兴趣的朋友可以参考本人的另一篇文章
Hadoop 综合揭秘——MapReduce 基础编程（介绍 Combine、Partitioner、WritableComparable、WritableComparator 使用方式）
里面会有更详细的介绍

  1 public class ExamineDetailKey implements WritableComparable<ExamineDetailKey> {
  2     //当TableType为0时是审核结果表数据，为1是审核流程记录表数据
  3     public static final IntWritable IsExamine=new IntWritable(0);
  4     public static final IntWritable IsAssess=new IntWritable(1); 
  5     public IntWritable TableType=new IntWritable();    
  6     public Text Id=new Text();
  7     
  8     @Override
  9     public void readFields(DataInput in) throws IOException {
 10         // TODO 自动生成的方法存根
 11         this.TableType.readFields(in);
 12         this.Id.readFields(in);
 13     }
 14 
 15     @Override
 16     public void write(DataOutput out) throws IOException {
 17         // TODO 自动生成的方法存根
 18         this.TableType.write(out);
 19         this.Id.write(out);
 20     }
 21 
 22     @Override
 23     public int compareTo(ExamineDetailKey o) {
 24         // TODO 自动生成的方法存根
 25         if(this.Id.equals(o.Id))
 26            return this.TableType.compareTo(o.TableType);
 27         else
 28            return this.Id.compareTo(o.Id);
 29     }
 30     
 31     @Override
 32     public boolean equals(Object o){
 33         if(!(o instanceof ExamineDetailKey))
 34             return false;
 35         
 36         ExamineDetailKey writable=(ExamineDetailKey)o;
 37         if(this.Id.equals(writable.Id)&&this.TableType.equals(writable.TableType))
 38             return true;
 39         else
 40             return false;
 41     }
 42     
 43     @Override
 44     public int hashCode(){
 45         return (this.Id.toString()+this.TableType.toString()).hashCode();
 46     }
 47 }
 48 
 49 public class ExamineDetailValue implements Writable{
 50     public Text Id=new Text();
 51     public Text Name=new Text();
 52     public IntWritable Age=new IntWritable();
 53     public Text Gender=new Text();
 54     public Text Area=new Text();
 55     public IntWritable Approved=new IntWritable(); 
 56     public Text Assessor=new Text();
 57     public Text ApprovedDate=new Text();
 58     public Text FilingDate=new Text();
 59     public Text Goods=new Text();
 60 
 61     @Override
 62     public void readFields(DataInput in) throws IOException {
 63         // TODO 自动生成的方法存根
 64         this.Id.readFields(in);
 65         this.Name.readFields(in);
 66         this.Age.readFields(in);
 67         this.Gender.readFields(in);
 68         this.Area.readFields(in);
 69         this.Approved.readFields(in);
 70         this.Assessor.readFields(in);
 71         this.ApprovedDate.readFields(in);
 72         this.FilingDate.readFields(in);
 73         this.Goods.readFields(in);
 74     }
 75 
 76     @Override
 77     public void write(DataOutput out) throws IOException {
 78         // TODO 自动生成的方法存根
 79         this.Id.write(out);
 80         this.Name.write(out);
 81         this.Age.write(out);
 82         this.Gender.write(out);
 83         this.Area.write(out);
 84         this.Approved.write(out);
 85         this.Assessor.write(out);
 86         this.ApprovedDate.write(out);
 87         this.FilingDate.write(out);
 88         this.Goods.write(out);
 89     }
 90 }
 91 
 92 public class ExamineDetailSortComparator extends WritableComparator {
 93      public ExamineDetailSortComparator(){
 94          super(ExamineDetailKey.class,true);
 95      }
 96 }
 97 
 98 public class ExamineDetailGroupComparator extends WritableComparator {
 99      public ExamineDetailGroupComparator(){
100          super(ExamineDetailKey.class,true);
101      }
102      
103       @Override
104      public int compare(WritableComparable a,WritableComparable b){
105          ExamineDetailKey key1=(ExamineDetailKey) a;
106          ExamineDetailKey key2=(ExamineDetailKey) b;
107          return key1.Id.compareTo(key2.Id);
108      }
109 }
110 
111 public class MrToHBaseExample extends Configured implements Tool{
112     
113     public static class MyMapper extends TableMapper<ExamineDetailKey,ExamineDetailValue>{
114         private String tablename;
115         
116         @Override
117         public void setup(Context context) throws IOException,InterruptedException{
118                 TableSplit split=(TableSplit)context.getInputSplit();
119                 tablename=new String(split.getTableName());
120         }
121         
122         public void map(ImmutableBytesWritable writable,Result result,Context context)
123             throws IOException,InterruptedException{
124             ExamineDetailKey key=new ExamineDetailKey();
125             ExamineDetailValue value=new ExamineDetailValue();
126             
127             for(Cell cell:result.listCells()){
128                 String data=Bytes.toString(CellUtil.cloneValue(cell));
129                 if(tablename.equals("Examine")){
130                     //设置表类型辨识符TableType
131                     key.TableType=ExamineDetailKey.IsExamine;
132                     //获取行数据
133                     if((Bytes.toString(CellUtil.cloneQualifier(cell))).equals("id")){
134                         key.Id=new Text(data);
135                         value.Id=new Text(data);
136                     }
137                     if((Bytes.toString(CellUtil.cloneQualifier(cell))).equals("name"))
138                         value.Name=new Text(data);
139                      if((Bytes.toString(CellUtil.cloneQualifier(cell))).equals("age"))
140                          value.Age=new IntWritable(new Integer(data));
141                     if((Bytes.toString(CellUtil.cloneQualifier(cell))).equals("gender"))
142                         value.Gender=new Text(data);
143                     if((Bytes.toString(CellUtil.cloneQualifier(cell))).equals("area"))
144                         value.Area=new Text(data);
145                      if((Bytes.toString(CellUtil.cloneQualifier(cell))).equals("approved"))
146                          value.Approved=new IntWritable(new Integer(data));
147                 }else if(tablename.equals("Assess")){
148                     //设置表类型辨识符TableType
149                     key.TableType=ExamineDetailKey.IsAssess;
150                     //获取行数据
151                     if((Bytes.toString(CellUtil.cloneQualifier(cell))).equals("id")){
152                         key.Id=new Text(data);
153                         value.Id=new Text(data);    
154                     }
155                     if((Bytes.toString(CellUtil.cloneQualifier(cell))).equals("assessor"))
156                         value.Assessor=new Text(data);
157                      if((Bytes.toString(CellUtil.cloneQualifier(cell))).equals("approvedDate"))
158                          value.ApprovedDate=new Text(data);
159                     if((Bytes.toString(CellUtil.cloneQualifier(cell))).equals("filingDate"))
160                         value.FilingDate=new Text(data);
161                     if((Bytes.toString(CellUtil.cloneQualifier(cell))).equals("goods"))
162                         value.Goods=new Text(data);
163                 }
164             }
165 
166             context.write(key,value);
167         }      
168     }
169     
170     public static class MyReducer extends Reducer<ExamineDetailKey,ExamineDetailValue,NullWritable,Text>{        
171         public void reduce(ExamineDetailKey key,Iterable<ExamineDetailValue> values,Context context)
172             throws IOException,InterruptedException{
173             String data=key.Id.toString();
174 
175             for(ExamineDetailValue value : values){
176                 if(key.TableType.equals(ExamineDetailKey.IsExamine)){
177                     data+=","+value.Name.toString()+","+value.Age.toString()+","
178                             +value.Gender.toString()+","+value.Area.toString()+","
179                             +value.Approved.toString();
180                 }else if(key.TableType.equals(ExamineDetailKey.IsAssess)){
181                     data+=","+value.Assessor.toString()+","
182                             +value.ApprovedDate.toString()+","+value.FilingDate.toString()+","
183                             +value.Goods.toString();
184                     context.write(NullWritable.get(),new Text(data));
185                 }
186             }
187         }
188     }
189 
190     public int run(String[] arg0) throws Exception {
191         // TODO 自动生成的方法存根
192          // TODO Auto-generated method stub
193         //生成表Examine,列簇family1
194         //HBaseUtils.createTable("Examine", "family1");
195         //建立任务Job
196         Job job=Job.getInstance(getConf());
197         job.setJarByClass(MrToHBaseExample.class);
198         //初始化TableMapper,把Examine表中的数据与Assess中的数据发送到Mapper
199         Scan scan1=new Scan();
200         scan1.addFamily(Bytes.toBytes("family1"));
201         scan1.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, "Examine".getBytes());
202         Scan scan2=new Scan();
203         scan2.addFamily(Bytes.toBytes("family1"));
204         scan2.setAttribute(Scan.SCAN_ATTRIBUTES_TABLE_NAME, "Assess".getBytes());
205         List<Scan> list=new ArrayList<Scan>();
206         list.add(scan1);
207         list.add(scan2);
208         TableMapReduceUtil.initTableMapperJob(list,MyMapper.class,ExamineDetailKey.class,ExamineDetailValue.class,job);
209         //注册Key/Value类型为Text
210         job.setOutputKeyClass(NullWritable.class);
211         job.setOutputValueClass(Text.class);
212         //TableMapReduceUtil.initTableMapperJob方法中已经绑定Mapper输出类型，下面方法可忽略
213         //job.setMapOutputKeyClass(ExamineDetailKey.class);
214         //job.setMapOutputValueClass(ExamineDetailValue.class);
215         //注册Mapper及Reducer处理类
216         job.setMapperClass(MyMapper.class);
217         job.setReducerClass(MyReducer.class);
218         //设置SortComparator和GroupComarartor
219         job.setSortComparatorClass(ExamineDetailSortComparator.class);
220         job.setGroupingComparatorClass(ExamineDetailGroupComparator.class);
221         //输入输出数据格式化类型为TextInputFormat/TableOutputFormat
222         job.setInputFormatClass(MultiTableInputFormat.class);
223         job.setOutputFormatClass(TextOutputFormat.class);
224         //伪分布式情况下不设置时默认为1
225         //job.setNumReduceTasks(1);
226         //注册自定义Partitional类
227         //job.setPartitionerClass(MyPatitional.class);
228         //获取命令参数
229         String[] args=new GenericOptionsParser(getConf(),arg0).getRemainingArgs();
230         FileOutputFormat.setOutputPath(job,new Path(args[0]));    
231         boolean status=job.waitForCompletion(true);
232         if(status)
233             return 0;
234         else
235             return 1;
236     }
237     
238     public static void main(String[] args) throws Exception{
239         Configuration conf=new Configuration();
240         ToolRunner.run(new MrToHBaseExample(), args);      
241     } 
242 }

运行结果

 返回目录

 

 

本章小结

本文的主要目的是介绍 HBase 的运行原理，介绍 HBase 的常用 Java API 开发实例，讲解 HBase 与 MR 之间的关系。在现今大数据年代，了解NoSQL的开发可以说是技术人员入门的必修课程，希望本文对各位的工作学习有所帮助。
由于本人时间紧迫，文章中有所缺漏的地方敬请点评。

对 JAVA 开发有兴趣的朋友欢迎加入QQ群：174850571 共同探讨！
对 .NET  开发有兴趣的朋友欢迎加入QQ群：162338858 共同探讨 ！

 

Hadoop 综合揭秘

HBase 的原理与应用

MapReduce 基础编程（介绍 Combine、Partitioner、WritableComparable、WritableComparator 使用方式）

 

作者：风尘浪子

https://www.cnblogs.com/leslies2/p/9530378.html

原创作品，转载时请注明作者及出处