高级用法

一，行键设计：

HBase有两种基本的键结构：行键(rowkey)和列键(columnkey)。两者都可以存储有意义的信息，这些信息有两类，一种是键本身存储的内容，另一种是键的排列顺序。

HBase 的表中数据分隔主要是使用列族（column family）而不是列，并非列式存储，如下图，表示了用户逻辑上把一个单元格的数据存到了一张表中，实际上底层存储是按列族线性地存储单元格，并包含了一些必要信息：

左上图是数据的逻辑视图，列键包含了列族和列（cf1:c1、cf2:c1...），每一行必须有一个行键（r1、r2...），可以通过行键获取到一行的所有列。
右上图表示逻辑视图转换为实际物理存储，每一行的每个单元格被有序存储，不同列族的数据存储在不同的文件（store file）中，同一个列族的数据存储在一个文件中（实际上会有多个 HFile，逻辑上属于同一个）。并且单元格的每个历史版本也会在存储文件中存在。
HBase 不存储 NULL 值，所以表可以很稀疏，没有值的单元格不会占用存储空间。每个单元格在存储文件中都存储了该记录在表中所处位置的相关信息。
如右下图所示，每一个单元格都包含：行键、列族、列键、写入时间戳和列值。并且按照时间戳降序排列。在读取 HFile 时最新的值会先被读到。单元格在 HBase 中被叫做 KeyValue，存储时先按行键排序，再按列键排序。因此用户可以按行键、或按行键的范围指定查询，同时指定列族，可以有效的减少检索的文件。
列和时间戳也可以用来筛选数据。查询数据时可以指定返回的列（可以配合过滤器 Filter 实现比较复杂的筛选），也可以指定时间范围进行过滤。
列值是最后一个筛选条件，与列的筛选类似，对每一个 KeyValue 都要经过过滤器检查，对性能的提升不明显，会减少返回给客户端的数据量。

二，高表与宽表的选择：

hbase中的宽表是指很多列较少行，即列多行少的表，一行中的数据量较大，行数少；高表是指很多行较少列，即行多列少，一行中的数据量较少，行数大。

hbase的row key是分布式的索引，也是分片的依据。
hbase的row key + column family + column qualifier + timestamp + value 是HFile中数据排列依据。HFile据此，对数据的索引到data block级别，而不是行级别。所以这种key是HFile内部的粗粒度（data block粒度）本地索引的主键。

据此，在HBase中使用宽表、高表的优劣总结如下：

• 查询性能：高表更好，因为查询条件都在row key中, 是全局分布式索引的一部分。高表一行中的数据较少。所以查询缓存BlockCache能缓存更多的行，以行数为单位的吞吐量会更高。
• 分片能力：高表分片粒度更细，各个分片的大小更均衡。因为高表一行的数据较少，宽表一行的数据较多。HBase按行来分片。
• 元数据开销：高表元数据开销更大。高表行多，row key多，可能造成region数量也多，- root -、 .meta表数据量更大。过大的元数据开销，可能引起HBase集群的不稳定、master更大的负担（这方面后续再好好总结）。
• 事务能力：宽表事务性更好。HBase对一行的写入（Put）是有事务原子性的，一行的所有列要么全部写入成功，要么全部没有写入。但是多行的更新之间没有事务性保证。
• 数据压缩比：如果我们对一行内的数据进行压缩，宽表能获得更高的压缩比。因为宽表中，一行的数据量较大，往往存在更多相似的二进制字节，有利于提高压缩比。通过压缩，缓解了宽表一行数据量太大，并导致分片大小不均匀的问题。查询时，我们根据row key找到压缩后的数据，进行解压缩。而且解压缩可以通过协处理器（coproesssor）在HBase服务器上做，而不是在业务应用的服务器上做，以充分应用HBase集群的CPU能力。

三，分页：

offset和limit结合“个人觉得这个和普通分页思路基本是差不太多，算不上是高级功能吧”

具体步骤：

1. 在起始键的位置处打开一个扫描器。
2. 跳过offset数目的行。
3. 读取limit数目的行，并返回给上层应用。
4. 关闭扫描

四，高级模式：

• 跨语言的IDL(interface definition language)
• avro
• protocol buffers

HAvroBase是一个使用Avro在每列中存储复杂记录的典型项目。它使用Avro的接口定义语言（IDL）来定义实际的模式。这个模式被用于在表的任意列中存储Avro形式序列化的记录。

Protocal buffers

和Avro类似，用户可以用Protocol Buffers的IDL来定义一个外部模式，这个模式被用来序列化复杂的数据结构到HBase中。

这种方法本质上是为用户提供一种可以定义初始模式的语言，然后用户可以通过增加或者删除字段来更新这个模式。序列化API能够使用新模式来读取旧模式，缺失的字段被忽略或者用默认值来填充。

五，辅助索引：

尽管HBase没有为辅助索引提供原生支持，但是有些应用场景仍需要使用辅助索引。通常的需求是用户能够通过主坐标（行键，列族和列限定符）来查找一个单元格，也可以通过一个其他类型的坐标来进行查找。此外，用户可以按照辅助索引的顺序从主表扫描数据。

与关系数据库系统中的索引类似，辅助索引存储了一个新坐标和现有的坐标之间的映射关系。看下可行的解决方案：

1，把责任完全转移到应用层的典型做法是一个数据表和一个（或者多个）查找/映射表结合起来。每当程序写数据表时，它也同时更新映射表（也被成为辅助索引表）。读数据时可以直接在主表中进行查询，从辅助索引表中先查找原表的行键，再在原表中读取数据。

优点：整个逻辑代码都是由客户端处理，用户可以按照需求设计映射关系。

缺点：HBase不能保障跨行操作原子性；缺少事物支持可能导致数据被存储在数据表中，但是在辅助索引表中没有相应映射。

2，大多数客户端和服务端的类都被添加了索引支持功能的类替换掉了。客户端管理索引在单独表中存储主键和辅助键之间的映射关系，与之不同的是，这些操作在ITHBase中被处理，同时主表与辅助索引表之间的关系由相应的描述符（descriper）定义。结合事物支持的索引更新，这个方案提供了HBase辅助索引的完全实现。

缺点：不支持最新HBase版本；增加了同步开销，导致性能下滑；

六，搜索继承

经常使用索引用户可以按照行键以外的顺序来遍历数据表。但用户任然受限于使用键或过滤器来筛选数据，或者直接遍历数据来查找所需的内容。一个非常普遍的需求是使用任意关键字来搜索数据，满足这种需求往往需要集成一个完整的搜索引擎。

常见的做法是基于 Apache Lucene 的解决方案，例如 Lucene或者Solr或者ES等。但是可以通过以下一系列可行方法。

客户端管理

FackBook的解决方案是，收件箱搜索系统（Facebook inbox search），具体模式如下：

• 每一行是一个单独的收件箱，即每个用户在搜索表中都有一个单独的行。
• 列是消息中被索引的词语
• 版本是消息ID
• 值包括附加信息，例如，词组在文件中的位置

这个模式使用户能很容易在收件箱中搜索包含特定写关键词的消息。布尔运算符，例如and和or，可以在客户端逻辑中实现，并用来归并找到的文件列表。用户也可以有效的实现输入提示查询，用户可以开始键入一个词，搜索将找到所有包含以用户输入作为前缀的词的消息。

关于“收件箱搜素系统”特意查阅相关资料。《Cassandra原理》

Facebook开发Cassandra初衷是用于Inbox Search，但是后来的Message System则使用了HBase，Facebook对此给出的解释是Cassandra的最终一致性模型不适合Message System，HBase具有更简单的一致性模型，当然还有其他的原因。HBase更加的成熟，成功的案例也比较多等等。Twitter和Digg都曾经很高调的选用Cassandra，但是最后也都放弃了，当然Twitter还有部分项目也还在使用Cassandra，但是主要的Tweet已经不是了。

也就是说因为HBase案例非常成熟，以至于Facebook放弃了Cassandra。而2007 年 Facebook 为了解决消息收件箱搜索问题（ Inbox Search problem）而开始设计 Cassandra 项目。

OK感受到了HBase强大力量。

七，事务

好吧，看到HBase关于事务的处理时候，发现了传统数据库的缩影。当然HBase本身并不是和传统一样的事务，这种事务是一种逻辑上的事务。

通常情况下，用户是用不到事务的，因为范式化的数据模型能够被转化成单个表的存储，所以此时不需要分布式事务的支持，同时也不需要为保证一致性而产生额外的开销。但是这里有两种方案HBase事务

1，事务型HBase

带索引的事务型HBase（Indexed Transactional HBase）项目有一些取代默认客户端类和服务端类的拓展类，它们增加了跨行甚至跨表的事务支持。在region服务中，更准确的说，每个region都保持了一个事务列表。该列表是由beginTransaction()调用初始化，并且由相应的commit()调用结束。每次读写操作都有一个事务ID，以保护调用不受其他调用的影响。

2，Zookeeper

HBase运行时需要一个Zookeeper集群，它在集群的启动中扮演着种子或引导启动的角色。有一些可用的模板和方法展示了如何将Zookeeper作为一个事务来控制后端。例如，Cages“笼子的意思，貌似zookeeper又是动物园，这两者之间关系定义的非常有趣形象”（http://code.google.com/p/cagesl）项目提供了一个锁的抽象概念，即跨多个资源的锁，并计划增加一个专门的事务来把Zookeeper当做分布式协调系统使用。

Zookeeper也提供了一个能够被用于实现两阶段提交协议的锁的方案。它使用一个特定的znode来代表事务，并且每个参与客户端对应一个孩子znode。客户端可以使用自己的znode标志自己在事务中的那部分是成功还是失败的。其他客户端可以监控同级的znode，并采适当的行动。

八，布隆过滤器

布隆过滤器（Bloom Filter）是1970由布隆（不是LOL中的那个辅助）提出的。通过一个很长的二进制向量于一系列随即哈希函数生成。

何为布隆过滤器？

首先，我们应当知道，hash是内存中使用的经典数据结构。

当我们需要判读一个元素是否在一个集合当中时，我们可以用哈希表来判断。在集合较小的情况下，hash是可行而且高效的。

然而数据量以PT计的大数据场景中，很多时候，hash便力有未逮。这是因为在海量数据下hash要占据巨额内存空间，这远远超过我们能够提供的内存大小。

例如在黑名单过滤当中，我们有100亿的网站黑名单url需要过滤，假设一个url是64bytes。如果我们用hash表来做，那么我们至少需要6400亿字节即640G的内存空间（实际所需空间还远大于此），空间消耗巨大，必须要多个服务器来同时分摊内存。

然而我们是否能用更加精简的结构来做这件事呢？布隆过滤器就是这样一个高度节省空间的结构，并且其时间也远超一般算法，但是布隆过滤器存在一定的失误率，例如在网页URL黑名单过滤中，布隆过滤器绝不会将黑名单中网页查错，但是有可能将正常的网页URL判定为黑名单当中的，它的失误可以说是宁可错杀，不可放过。不过布隆过滤器的失误率可以调节，下面我们会详细介绍。

布隆过滤器实际就是一种集合。假设我们有一个数组，它的长度为L，从0-（L-1）位置上，存储的不是一个字符串或者整数，而是一个bit，这意味它的取值只能为0或1.

例如我们实现如下的一个数组：

int[] array = new int[1000];

该数组中有1000个int类型的元素，而每一个int由有4个byte组成，一个byte又由8个bit组成，所以一个int就由32个bit所组成。

所以我们申请含1000整数类型的数组，它就包含32000个bit位。

但是我们如果想将第20001个bit位描黑，将其改为1，我们需要怎样做呢。

首先我们的需要定位，这第20001个bit位于哪个整数，接着我们需要定位该bit位于该整数的第几个bit位。

int index = 20001;
int intIndex = index/32;
int bitIndex = index%32;

然后我们就将其描黑：

array[intIndex] =(array[intIndex] | (1 << bitIndex));

这段代码可能有些同学不能理解，下面我们详细解释一下。我们的数组是0-999的整型数组，但是每个位置上实际保存的是32个bit，我们的intIndex就是代表第几个整数，定位到625，而bitIndex定位到第625个整数中的第几个bit，为1。

所以，我们需要描黑的位置是第625号元素的第1个bit。

而(1 << bitIndex)代表将1左移到1位置，即是代表只有1位置为1，而其他位置为零。就相当于获得了这样一串二进制数：

00000000 00000000 00000000 00000001

然后我们将这样的二进制数与array[indIndex]进行逻辑或运算，就能将第625号元素的第一个bit描黑变1，最后我们将描黑后的元素赋值给array[indIndex]，完成运算。

这里我们采用的是int类型的数组，如果我们想更加节省空间，我们就能创建long类型的数组，这样申请1000个数组空间，我们就能得到64000个bit。

然后我们还能进一步扩展，将数组做成矩阵：

long[][] map = new long[1000][1000];

有了这些基础以后，我们如何设计黑名单问题呢？

假设我们已经有了一个拥有m个bit的数组，然后我们将一个URL通过哈希函数计算出hashcode，然后hashcode%m将对应位置描黑。

然而这还不是真正的布隆过滤器，真正的布隆过滤器是通过多个哈希函数对一个URL进行计算，得到hashcode，然后在对不同位置的bit进行描黑。

注意：布隆过滤器采用的多个哈希函数必须是相互独立的，前面我们已经介绍了如何通过一个哈希函数构造多个独立哈希函数的方法。

当我们将一个URL通过n个哈希函数，得到hashcode，模以m，再将对应位置描黑之后。我们可以说该URL已经进入我们的布隆过滤器当中了。

接下来，我们将黑名单中所有URL都通过哈希函数，进入布隆过滤器中（布隆过滤器并不真正存储实际的URL）。

对于一个新的URL，我们要查询其是否在黑名单中，我们便通过同样的n个哈希函数，计算出n个位置，然后我们查询这n个位置是否都被描黑，如果都被描黑，我们就说该URL在黑名单当中。如果n个位置但凡有一个不为黑，我们就说该URL不在该黑名单中。

注意：我们的数组不应该过小，不然很可能数组中的大多数位置都被描黑，这很容易将正常的URL判定为不合法的，这也是布隆过滤器的失误率来源。

算法解析

1、BloomFilter 流程

　　1. 首先需要 k 个 hash 函数，每个函数可以把 key 散列成为 1 个整数；

　　2. 初始化时，需要一个长度为 n 比特的数组，每个比特位初始化为 0；

　　3. 某个 key 加入集合时，用 k 个 hash 函数计算出 k 个散列值，并把数组中对应的比特位置为 1；

　　4. 判断某个 key 是否在集合时，用 k 个 hash 函数计算出 k 个散列值，并查询数组中对应的比特位，如果所有的比特位都是1，认为在集合中。

2、关于哈希冲突

　　假设 Hash 函数是良好的，如果我们的位阵列长度为 m 个点，那么如果我们想将冲突率降低到例如 1%, 这个散列表就只能容纳 m/100个元素。显然这就不叫空间效率了（Space-efficient）了。解决方法，就是使用多个 Hash，如果它们有一个说元素不在集合中，那肯定就不在。如果它们都说在，虽然也有一定可能性它们都在说谎，不过直觉上判断这种事情的概率是比较低的。--- 如上 BloomFilter 流程

　　一个 Bloom Filter 是基于一个 m 位的位向量（b1,…bm），这些位向量的初始值为0。另外，还有一系列的hash函数（h1,…,hk），这些 hash 函数的值域属于1~m。

解析：m=18，k=3；插入 x 是，三个 hash 函数分别得到蓝线对应的三个值，并将对应的位向量改为1，插入 y，z 时，类似的，分别将红线，紫线对应的位向量改为1。查找时，当查找 x 时，三个 hash 值对应的位向量都为1，因此判断 x 在此数据集中。y，z 也是如此。但是查找 w 时，w 有个 hash 值对应的位向量为0，因此可以判断不在此集合中。但是，假如 w 的最后那个 hash 值是1，这时就会认为 w 在此集合中，而事实上，w 可能不在此集合中，因此可能出现误报。显然的，插入数据越多，1的位数越多，误报的概率越大。

　　Wiki的Bloom Filter词条有关于误报的概率的详细分析：Probability of false positives。从分析可以看出，当 k 比较大时，误报概率还是比较小的。

hbase中的布隆过滤器

布隆过滤器是hbase中的高级功能，它能够减少特定访问模式（get/scan）下的查询时间。不过由于这种模式增加了内存和存储的负担，所以被默认为关闭状态。

hbase支持如下类型的布隆过滤器：

1、NONE 不使用布隆过滤器

2、ROW 行键使用布隆过滤器

3、ROWCOL 列键使用布隆过滤器

其中ROWCOL是粒度更细的模式。

在介绍为什么hbase要引入布隆过滤器之前，我们先来了解一下hbase存储文件HFile的块索引机制

我们知道hbase的实际存储结构是HFile，它是位于hdfs系统中的，也就是在磁盘中。而加载到内存中的数据存储在MemStore中，当MemStore中的数据达到一定数量时，它会将数据存入HFile中。

HFIle是由一个个数据块与索引块组成，他们通常默认为64KB。hbase是通过块索引来访问这些数据块的。而索引是由每个数据块的第一行数据的rowkey组成的。当hbase打开一个HFile时，块索引信息会优先加载到内存当中。

然后hbase会通过这些块索引来查询数据。

但是块索引是相当粗粒度的，我们可以简单计算一下。假设一个行占100bytes的空间，所以一个数据块64KB，所包含的行大概有：(64 * 1024)/100 = 655.53 = ~700行。而我们只能从索引给出的一个数据块的起始行开始查询。

如果用户随机查找一个行键，则这个行键很可能位于两个开始键（即索引）之间的位置。对于hbase来说，它判断这个行键是否真实存在的唯一方法就是加载这个数据块，并且扫描它是否包含这个键。

同时，还存在很多情况使得这种情况更加复杂。

对于一个应用来说，用户通常会以一定的速率进行更新数据，这就将导致内存中的数据被刷写到磁盘中，并且之后系统会将他们合并成更大的存储文件。在hbase的合并存储文件的时候，它仅仅会合并最近几个存储文件，直至合并的存储文件到达配置的最大大小。最终系统中会有很多的存储文件，所有的存储文件都是候选文件，其可能包含用户请求行键的单元格。

我们可以看到，这些不同的文件都来着同一个列族，所以他们的行键分布类似。所以，虽然我们要查询更新的特定行只在某个或者某几个文件中，但是采用块索引方式，还是会覆盖整个行键范围。当块索引确定那些块可能含有某个行键后，regionServer需要加载每一个块来检查该块中是否真的包含该行的单元格。

这里有几点细节：

1、布隆过滤器的存储在哪？

对于hbase而言，当我们选择采用布隆过滤器之后，HBase会在生成StoreFile（HFile）时包含一份布隆过滤器结构的数据，称其为MetaBlock；MetaBlock与DataBlock（真实的KeyValue数据）一起由LRUBlockCache维护。所以，开启bloomfilter会有一定的存储及内存cache开销。但是在大多数情况下，这些负担相对于布隆过滤器带来的好处是可以接受的。

2、采用布隆过滤器后，hbase如何get数据？

在读取数据时，hbase会首先在布隆过滤器中查询，根据布隆过滤器的结果，再在MemStore中查询，最后再在对应的HFile中查询。

3、采用ROW还是ROWCOL布隆过滤器？

这取决于用户的使用模式。如果用户只做行扫描，使用更加细粒度的行加列布隆过滤器不会有任何的帮助，这种场景就应该使用行级布隆过滤器。当用户不能批量更新特定的一行，并且最后的使用存储文件都含有改行的一部分时，行加列级的布隆过滤器更加有用。

例如：ROW 使用场景假设有2个Hfile文件hf1和hf2， hf1包含kv1（r1 cf:q1 v）、kv2（r2 cf:q1 v） hf2包含kv3（r3 cf:q1 v）、kv4（r4 cf:q1 v） 如果设置了CF属性中的bloomfilter（布隆过滤器）为ROW，那么get(r1)时就会过滤hf2，get(r3)就会过滤hf1 。

ROWCOL使用场景假设有2个Hfile文件hf1和hf2， hf1包含kv1（r1 cf:q1 v）、kv2（r2 cf:q1 v） hf2包含kv3（r1 cf:q2 v）、kv4（r2 cf:q2 v） 如果设置了CF属性中的bloomfilter为ROW，无论get(r1,q1)还是get(r1,q2)，都会读取hf1+hf2；而如果设置了CF属性中的bloomfilter为ROWCOL，那么get(r1,q1)就会过滤hf2，get(r1,q2)就会过滤hf1。

注意：ROW和ROWCOL只是名字上有联系，但是ROWCOL并不是ROW的扩展，也不能取代ROW

九，版本管理

隐式版本控制：

在用户确保具有服务器上的时钟是同步的之前，有些问题就已经被指出了。其中一个可能出现的问题是，当用户跨服务器把数据存储在多行中时，使用隐式的时间戳可能让用户最终得到完全不同的时间集。

例如，假设用户使用HBase的URL短网址服务，并且为现有用户存储了3个新缩短过的URL。所有的键都被认为是完全分布式的，所以3个新行最后在不同的region服务器上，进一步假设这些服务器时间都相隔一个小时，如果用户通过扫描客户端来获取最近一个小时的更新的缩写的URL列表，用户将会遗漏一些数据，因为它们被保存时的时间戳和客户端的时间戳相差超过一个小时。

用户可以通过在存储这些数据时候设定商定的或者共享的时间戳来避免这个问题。put操作是允许用户设置一个客户端的时间戳作为代替，并以此覆盖覆盖服务器时间戳。显然依靠服务器来完成这个是不方便的，但是一些情况下用户可能还需要使用这个方法。

region爆裂也会引起类似问题存在。假设用户把一个值存储在一个比集群中所有其他服务器时间都超前一个小时服务器上，并且使用服务器隐式时间戳，十分钟后这个region被拆分，并且用户一半的数据跟新被移到了另一个服务器上。5分钟后，当用户再向同样的列中插入一个新值的，服务器会自动添加时间戳。现在新移植被认为比之前添加的数据更老，因为第一个版本的时间戳比当前服务器时间超前一个小时，此时如果用户使用标准的get方法检索这个数据将会得到第一个之前存储的值。

归根到底还是服务器时间戳不一致，导致了版本时间戳查询出现了误差。

自定义版本控制

由于你可以指定自己的时间戳，当然也可以创建自己的版本控制计划。在覆盖服务端基于同步的服务器时间的时间戳时，用户可以不使用基于时间的版本。

例如，用户可以吧时间戳和一个全局的数字生成器结合起来使用，这个生成器提供了从“1”开始一直增加的顺序数字。每次用户插入一个新值时，就会获得一个新数字，并且在调用put函数时会用到它。

用户必须为每个put操作都这么做，否则服务器将插入一个机遇服务器端的时间的时间戳。表或者列的描述中又一个标志可以表明你使用的是自定义的时间戳（即自定义的版本控制策略）。如果用户没有设置这个值，那么它将在后台被服务器端的时间戳替换掉。

当使用自己的时间戳值时，用户需要确保彻底的测试过自己的解决方案，因为这种方法并没有被广泛的用于生产中。
注意，负值的时间戳也未经测试过。尽管它在HBase开发圈中被讨论过几次，单它从未有被确认过能够正常的工作。
确保同一个单元的两次单独的更新操作不会使用相同的时间戳，这样会造成冲突。通常最后保存的值是可见的。

记录ID：FaceBook的（FaceBook inbox search）中，它使用时间戳值存储消息ID。由于这些ID随着时间而增长，并且在HBase中，版本的隐式排列顺序是降序排序。所以用户可以按时间排序获取匹配搜索列的最后10个版本以得到最近的10条信息。

数字生成器：生成器和基于时间戳的作用看起来是相同的。按照一个单调递增的数升序排列所有存储的值。不过它们的区别更微妙，因为JAVA计时器使用的精度是毫秒，这意味着使用完全相同的时间存储两个值不太可能的，单并不是完全不可能的。如果用户需要一个全局完全唯一的版本控制解决方案，依赖数字生成器解决这个问题。

客户端API高级用法

一，过滤器

Filter（前面提到的高级用法中的布隆过滤器属于其中特殊一种）是任何一款数据存储架构的必备功能，HBase也不意外的拥有自己Filter功能。

先看看什么是顾虑器

HBase中有两种主要的获取数据的方式：get（）个scan（），它们都支持直接访问数据并通过指定起止行键访问数据的功能。用户可以添加非常多样化的限制条件来减少查询得到的数据。

当然用户也是可以通过继承Filter类来实现自己自定义的需求的。所有的过滤器都在服务端生效，叫做谓词下推（predicate push down）。这样可以保证被过滤掉的数据不会被传送到客户端。

过滤器层次结构

过滤器层次结构的最底层是Filter接口和FilterHbase抽象类，它们实现了过滤器的空壳和骨架，（个人在代码架构很喜欢这种结构层次思路，非常值得借鉴）这使得实际的过滤器类可以避免许多重复的结构代码。

大部分的Filter类都是继承了FilterHbase，也有一些间接的继承自该类。不过它们大部分使用流程是相同的。用户定义一个所有需要过滤的实例，同时把定义好的过滤器实例传递给Get（）或者Scan（）实例：

setFilter(filter)

当然实例化过程有需要很多参数，其中有一组特殊的过滤器它们是继承CompareFilter（）。用户需要提供至少两个特定的参数，这两个参数会被基类用户执行它的任务。

HBase过滤器的分类

比较过滤器

1、行键过滤器 RowFilter

Filter rowFilter = new RowFilter(CompareOp.GREATER, new BinaryComparator("95007".getBytes()));
scan.setFilter(rowFilter);

public class HbaseFilterTest {

    private static final String ZK_CONNECT_KEY = "hbase.zookeeper.quorum";
    private static final String ZK_CONNECT_VALUE = "hadoop1:2181,hadoop2:2181,hadoop3:2181";

    private static Connection conn = null;
    private static Admin admin = null;
    
    public static void main(String[] args) throws Exception {
        
        Configuration conf = HBaseConfiguration.create();
        conf.set(ZK_CONNECT_KEY, ZK_CONNECT_VALUE);
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
        Table table = conn.getTable(TableName.valueOf("student"));
        
        Scan scan = new Scan();
        
        Filter rowFilter = new RowFilter(CompareOp.GREATER, new BinaryComparator("95007".getBytes()));
        scan.setFilter(rowFilter);
        ResultScanner resultScanner = table.getScanner(scan);
        for(Result result : resultScanner) {
            List<Cell> cells = result.listCells();
            for(Cell cell : cells) {
                System.out.println(cell);
            }
        }
        
        
    }

2、列簇过滤器 FamilyFilter

Filter familyFilter = new FamilyFilter(CompareOp.EQUAL, new BinaryComparator("info".getBytes()));
scan.setFilter(familyFilter);
public class HbaseFilterTest {

    private static final String ZK_CONNECT_KEY = "hbase.zookeeper.quorum";
    private static final String ZK_CONNECT_VALUE = "hadoop1:2181,hadoop2:2181,hadoop3:2181";

    private static Connection conn = null;
    private static Admin admin = null;
    
    public static void main(String[] args) throws Exception {
        
        Configuration conf = HBaseConfiguration.create();
        conf.set(ZK_CONNECT_KEY, ZK_CONNECT_VALUE);
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
        Table table = conn.getTable(TableName.valueOf("student"));
        
        Scan scan = new Scan();

        Filter familyFilter = new FamilyFilter(CompareOp.EQUAL, new BinaryComparator("info".getBytes()));
        scan.setFilter(familyFilter);
        ResultScanner resultScanner = table.getScanner(scan);
        for(Result result : resultScanner) {
            List<Cell> cells = result.listCells();
            for(Cell cell : cells) {
                System.out.println(cell);
            }
        }
        
        
    }
    
    
}

3、列过滤器 QualifierFilter

Filter qualifierFilter = new QualifierFilter(CompareOp.EQUAL, new BinaryComparator("name".getBytes()));
scan.setFilter(qualifierFilter);
public class HbaseFilterTest {

    private static final String ZK_CONNECT_KEY = "hbase.zookeeper.quorum";
    private static final String ZK_CONNECT_VALUE = "hadoop1:2181,hadoop2:2181,hadoop3:2181";

    private static Connection conn = null;
    private static Admin admin = null;
    
    public static void main(String[] args) throws Exception {
        
        Configuration conf = HBaseConfiguration.create();
        conf.set(ZK_CONNECT_KEY, ZK_CONNECT_VALUE);
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
        Table table = conn.getTable(TableName.valueOf("student"));
        
        Scan scan = new Scan();
        
        Filter qualifierFilter = new QualifierFilter(CompareOp.EQUAL, new BinaryComparator("name".getBytes()));
        scan.setFilter(qualifierFilter);
        ResultScanner resultScanner = table.getScanner(scan);
        for(Result result : resultScanner) {
            List<Cell> cells = result.listCells();
            for(Cell cell : cells) {
                System.out.println(cell);
            }
        }
        
        
    }
    
    
}

4、值过滤器 ValueFilter

Filter valueFilter = new ValueFilter(CompareOp.EQUAL, new SubstringComparator("男"));
scan.setFilter(valueFilter);
public class HbaseFilterTest {

    private static final String ZK_CONNECT_KEY = "hbase.zookeeper.quorum";
    private static final String ZK_CONNECT_VALUE = "hadoop1:2181,hadoop2:2181,hadoop3:2181";

    private static Connection conn = null;
    private static Admin admin = null;
    
    public static void main(String[] args) throws Exception {
        
        Configuration conf = HBaseConfiguration.create();
        conf.set(ZK_CONNECT_KEY, ZK_CONNECT_VALUE);
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
        Table table = conn.getTable(TableName.valueOf("student"));
        
        Scan scan = new Scan();
        
        Filter valueFilter = new ValueFilter(CompareOp.EQUAL, new SubstringComparator("男"));
        scan.setFilter(valueFilter);
        ResultScanner resultScanner = table.getScanner(scan);
        for(Result result : resultScanner) {
            List<Cell> cells = result.listCells();
            for(Cell cell : cells) {
                System.out.println(cell);
            }
        }
        
        
    }
    
    
}

5、时间戳过滤器 TimestampsFilter

List<Long> list = new ArrayList<>();
list.add(1522469029503l);
TimestampsFilter timestampsFilter = new TimestampsFilter(list);
scan.setFilter(timestampsFilter);
public class HbaseFilterTest {

    private static final String ZK_CONNECT_KEY = "hbase.zookeeper.quorum";
    private static final String ZK_CONNECT_VALUE = "hadoop1:2181,hadoop2:2181,hadoop3:2181";

    private static Connection conn = null;
    private static Admin admin = null;
    
    public static void main(String[] args) throws Exception {
        
        Configuration conf = HBaseConfiguration.create();
        conf.set(ZK_CONNECT_KEY, ZK_CONNECT_VALUE);
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
        Table table = conn.getTable(TableName.valueOf("student"));
        
        Scan scan = new Scan();
        
        List<Long> list = new ArrayList<>();
        list.add(1522469029503l);
        TimestampsFilter timestampsFilter = new TimestampsFilter(list);
        scan.setFilter(timestampsFilter);
        ResultScanner resultScanner = table.getScanner(scan);
        for(Result result : resultScanner) {
            List<Cell> cells = result.listCells();
            for(Cell cell : cells) {
                System.out.println(Bytes.toString(cell.getRow()) + "\t" + Bytes.toString(cell.getFamily()) + "\t" + Bytes.toString(cell.getQualifier())
                + "\t" + Bytes.toString(cell.getValue()) + "\t" + cell.getTimestamp());
            }
        }
        
        
    }
    
    
}

专用过滤器

1、单列值过滤器 SingleColumnValueFilter ----会返回满足条件的整行

SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                "info".getBytes(), //列簇
                "name".getBytes(), //列
                CompareOp.EQUAL, 
                new SubstringComparator("刘晨"));
//如果不设置为 true，则那些不包含指定 column 的行也会返回
singleColumnValueFilter.setFilterIfMissing(true);
scan.setFilter(singleColumnValueFilter);

public class HbaseFilterTest2 {

    private static final String ZK_CONNECT_KEY = "hbase.zookeeper.quorum";
    private static final String ZK_CONNECT_VALUE = "hadoop1:2181,hadoop2:2181,hadoop3:2181";

    private static Connection conn = null;
    private static Admin admin = null;
    
    public static void main(String[] args) throws Exception {
        
        Configuration conf = HBaseConfiguration.create();
        conf.set(ZK_CONNECT_KEY, ZK_CONNECT_VALUE);
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
        Table table = conn.getTable(TableName.valueOf("student"));
        
        Scan scan = new Scan();
        
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(
                "info".getBytes(), 
                "name".getBytes(), 
                CompareOp.EQUAL, 
                new SubstringComparator("刘晨"));
        singleColumnValueFilter.setFilterIfMissing(true);
        
        scan.setFilter(singleColumnValueFilter);
        ResultScanner resultScanner = table.getScanner(scan);
        for(Result result : resultScanner) {
            List<Cell> cells = result.listCells();
            for(Cell cell : cells) {
                System.out.println(Bytes.toString(cell.getRow()) + "\t" + Bytes.toString(cell.getFamily()) + "\t" + Bytes.toString(cell.getQualifier())
                + "\t" + Bytes.toString(cell.getValue()) + "\t" + cell.getTimestamp());
            }
        }
        
        
    }
    
    
}

2、单列值排除器 SingleColumnValueExcludeFilter

SingleColumnValueExcludeFilter singleColumnValueExcludeFilter = new SingleColumnValueExcludeFilter(
                "info".getBytes(), 
                "name".getBytes(), 
                CompareOp.EQUAL, 
                new SubstringComparator("刘晨"));
singleColumnValueExcludeFilter.setFilterIfMissing(true);
        
scan.setFilter(singleColumnValueExcludeFilter);

public class HbaseFilterTest2 {

    private static final String ZK_CONNECT_KEY = "hbase.zookeeper.quorum";
    private static final String ZK_CONNECT_VALUE = "hadoop1:2181,hadoop2:2181,hadoop3:2181";

    private static Connection conn = null;
    private static Admin admin = null;
    
    public static void main(String[] args) throws Exception {
        
        Configuration conf = HBaseConfiguration.create();
        conf.set(ZK_CONNECT_KEY, ZK_CONNECT_VALUE);
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
        Table table = conn.getTable(TableName.valueOf("student"));
        
        Scan scan = new Scan();
        
        SingleColumnValueExcludeFilter singleColumnValueExcludeFilter = new SingleColumnValueExcludeFilter(
                "info".getBytes(), 
                "name".getBytes(), 
                CompareOp.EQUAL, 
                new SubstringComparator("刘晨"));
        singleColumnValueExcludeFilter.setFilterIfMissing(true);
        
        scan.setFilter(singleColumnValueExcludeFilter);
        ResultScanner resultScanner = table.getScanner(scan);
        for(Result result : resultScanner) {
            List<Cell> cells = result.listCells();
            for(Cell cell : cells) {
                System.out.println(Bytes.toString(cell.getRow()) + "\t" + Bytes.toString(cell.getFamily()) + "\t" + Bytes.toString(cell.getQualifier())
                + "\t" + Bytes.toString(cell.getValue()) + "\t" + cell.getTimestamp());
            }
        }
        
        
    }
    
    
}

3、前缀过滤器 PrefixFilter----针对行键

PrefixFilter prefixFilter = new PrefixFilter("9501".getBytes());
        
scan.setFilter(prefixFilter);

public class HbaseFilterTest2 {

    private static final String ZK_CONNECT_KEY = "hbase.zookeeper.quorum";
    private static final String ZK_CONNECT_VALUE = "hadoop1:2181,hadoop2:2181,hadoop3:2181";

    private static Connection conn = null;
    private static Admin admin = null;
    
    public static void main(String[] args) throws Exception {
        
        Configuration conf = HBaseConfiguration.create();
        conf.set(ZK_CONNECT_KEY, ZK_CONNECT_VALUE);
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
        Table table = conn.getTable(TableName.valueOf("student"));
        
        Scan scan = new Scan();
        
        PrefixFilter prefixFilter = new PrefixFilter("9501".getBytes());
        
        scan.setFilter(prefixFilter);
        ResultScanner resultScanner = table.getScanner(scan);
        for(Result result : resultScanner) {
            List<Cell> cells = result.listCells();
            for(Cell cell : cells) {
                System.out.println(Bytes.toString(cell.getRow()) + "\t" + Bytes.toString(cell.getFamily()) + "\t" + Bytes.toString(cell.getQualifier())
                + "\t" + Bytes.toString(cell.getValue()) + "\t" + cell.getTimestamp());
            }
        }
        
        
    }
    
    
}

4、列前缀过滤器 ColumnPrefixFilter

ColumnPrefixFilter columnPrefixFilter = new ColumnPrefixFilter("name".getBytes());
        
scan.setFilter(columnPrefixFilter);

public class HbaseFilterTest2 {

    private static final String ZK_CONNECT_KEY = "hbase.zookeeper.quorum";
    private static final String ZK_CONNECT_VALUE = "hadoop1:2181,hadoop2:2181,hadoop3:2181";

    private static Connection conn = null;
    private static Admin admin = null;
    
    public static void main(String[] args) throws Exception {
        
        Configuration conf = HBaseConfiguration.create();
        conf.set(ZK_CONNECT_KEY, ZK_CONNECT_VALUE);
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
        Table table = conn.getTable(TableName.valueOf("student"));
        
        Scan scan = new Scan();
        
        ColumnPrefixFilter columnPrefixFilter = new ColumnPrefixFilter("name".getBytes());
        
        scan.setFilter(columnPrefixFilter);
        ResultScanner resultScanner = table.getScanner(scan);
        for(Result result : resultScanner) {
            List<Cell> cells = result.listCells();
            for(Cell cell : cells) {
                System.out.println(Bytes.toString(cell.getRow()) + "\t" + Bytes.toString(cell.getFamily()) + "\t" + Bytes.toString(cell.getQualifier())
                + "\t" + Bytes.toString(cell.getValue()) + "\t" + cell.getTimestamp());
            }
        }
        
        
    }
    
    
}

5、分页过滤器 PageFilter

二，计数器

• 计数器实质是一个列，一个计数器对应一个列
• 用create创建，默认起始为0，每次加1
• 用incr增加，用get_counter 获取

单计数器

只操作一个计数器，用户自行定义列

多计数器

• 使用HTable()的方法increment() 创建实例，可以为实例传入行键（Rowkey），
• 此行可以包含多个计数器，
• 计数器更新时有锁，读取时没有锁

三，协处理器

• 允许用户将自己的代码打包上传到服务器上，在服务器上执行region级别的操作，类是触发器
• jar包可以动态加载
• HBase的授权认证和审查就是基于协处理器的
• 包含两种类型： observer 和 endpoint

observer

这类协处理器与触发器（TRIGGER）类似：回调函数（也成为是这钩子函数，hook）在一些特定事件发生时被执行。这些事件包括一些用户产生的事件，也包括服务器端内部自动产生的事件。

协处理器框架提供的接口如下

• RegionObserver：用户可以用这种的处理器处理数据修改事件，它们与表的region联系紧密。
• MasterObserver：可以被用作管理或DDL类型的操作，这些是集群级事件。
• WALObserver：提供控制WAL的钩子函数
• Observer提供了一些设计好的回调函数，每个操作在集群服务器端都可以被调用。

endpoint

除了事件处理之外还需要将用户的自定义操作添加到服务器端。用户代码可以被部署到管理数据的服务器端，例如，做一些服务器端计算的工作。

endpoint通过添加一些远程调用来动态的拓展RPC协议。可以把它理解为与RDBMS中类似的存储过程。endpoint可以与observer的实现组合起来直接作用于服务器端的状态。

Coprocessor类

所有协处理器的类都必须实现这个接口。它定义了协处理器的基本约定，并使得框架本身的管理变得容易。

Coprocessor.Priority枚举类定义的优先级

SYSTEM：高优先级，定义最先被执行的协处理器

USER：定义其他的协处理器，按顺序执行

协处理器的优先级决定了执行的顺序：系统（SYSTEM）级协处理器在用户（USER）级协处理器之前执行。

在同一个优先级中还有一个序号(sequence number)的概念，用来维护协处理器的加载顺序。序号从0开始依次增加。

这个数字的作用并不大，但用户可以依靠它们来为同一优先级的协处理器排序：在同一优先级下，它们按照其序号递增的顺序执行，即定义了执行顺序。

在协处理器生命周期中，它们由框架管理。Coprocessor接口提供了以下两个方法

void start(CoprocessorEnvironment env) throws IOException;
void stop(CoprocessorEnvironment env) throws IOException;

这两个方法在协处理器开始和结束时被调用。CoprocessorEnvironment用来在协处理器的生命周期中保持其状态。协处理器实例一直被保存在提供的环境中。

CoprocessorEnvironment类提供的方法

协处理器应当只与提供给它们的环境进行交互。这样的好处是可以保证没有会被恶意代码用来破坏数据的后门。

注意：协处理器应当使用getTable方法访问表数据。注意这个方法实际上在默认的HTable类上添加了特定的安全措施。例如，协处理器不可以对一行数据加锁。

在协处理器实例的生命周期中，Coprocessor接口的start和stop方法会被框架隐式调用，处理过程中的每一步都有一个状态。

Coprocessor.State枚举类定义的状态

四，HtablePool

使用HTablePool的原因主要有两点：

1. 创建HTable实例是比较耗时的；
2. HTable实例不是线程安全的。

Pool

Pool仅仅是一个接口

该接口对应着三个实现类，实现方式均不同

(1) ReusablePool

该类继承自ConcurrentLinkedQueue，顾名思义，池子内部使用队列（Queue）实现，获取资源里从队列头部摘取（poll），使用完成后再将资源放入（add）队列的尾部（不能超过池子配额限制）

@Override
public R get() {
     return poll();
}
 
@Override
public R put(R resource) {
     if (size() < maxSize) {
          add(resource);
     }
     return null;
}

说明：资源池中的资源数量不会超过maxSize，但池子中的资源实例可能发生变化，当某一时刻池子中的资源已被耗尽（资源均处于外部使用状态，且尚未归还），这里如果有资源请求到来，get返回为Null，此时外部可能会立刻创建新的资源，但是和资源被释放归还的时刻是不定的，可能某一资源（不是新创建的资源）归还时发现资源池中的资源数量已达到配额限制，则该资源不能进入资源池（可能被丢弃回收），虽然资源实例发生变化，但不会影响资源池中的资源数目。

（2）RoundRobinPool

该类继承自CopyOnWriteArrayList，顾名思义，使用轮询方式访问列表中的资源

@Override
public R get() {
     if (size() < maxSize) {
          return null;
     }
     nextResource %= size();
     R resource = get(nextResource++);
     return resource;
}

（3）ThreadLocalPool

该类继承自ThreadLocal，顾名思义，该资源池与线程有着紧密的联系，资源池中的资源数目取决于使用线程往资源池中添加资源的线程数目，每个线程仅可以使用属于自己线程的那个资源。private static final Map<ThreadLocalPool<?>, AtomicInteger> poolSizes = new HashMap<ThreadLocalPool<?>, AtomicInteger>();
poolSizes 用来记录各个资源池实例中各自的资源数目（同一程序中可以有多个不同的资源池）。资源的获取与归还主要利用ThreadLocal中的get、set方法

@Override
public R put(R resource) {
     R previousResource = get();
     if (previousResource == null) {
          AtomicInteger poolSize = poolSizes.get(this);
          if (poolSize == null) {
               poolSizes.put(this, poolSize = new AtomicInteger());
          }
          poolSize.incrementAndGet();
      }
      this.set(resource);
      return previousResource;
}

get、set中间的代码个人认为仅仅完成资源池审计功能。

PoolMap

HTablePool的资源池设计是以表名称（String、byte[]）为单位的，即对HBase中的每一张表维护着各自的连接池，因此在Pool之上有了PoolMap，实际使用中PoolMap的Key即为表名称。

但是Hbase1.1.2的HTablePool已经被弃用，准确而言不是被弃用 是不建议client用 都成了developers api

org.apache.hadoop.hbase.client.Connection

HConnection connection = HConnectionManager.createConnection(Configuration);
HTableInterface table = connection.getTable("myTable");
// use table as needed, the table returned is lightweight
table.close();
// use the connection for other access to the cluster
connection.close();

五，连接管理

每个HTable实例都需要建立和远程主机的连接。这些连接在内部使用HConnection类表示，更重要的是，其被HConnectionManager管理并共享。用户没必要同时和这两个类打交道，只需要创建一个Configuration实例，然后利用客户端API使用这些类。

Hbase内部使用键值映射来存储连接，使用Configuration实例作为键值映射的键。换句话说，当你创建很多HTable实例时，如果你提供了相同的Configuration引用，那么它们共享底层的同一个HConnection实例。

共享ZooKeeper实例

因为每个客户端最终都需要ZooKeeper连接来完成表的region初始寻址，连接一旦建立之后，共享就变的很有意义，这使得之后的客户端实例可以共用。

缓存通用资源

对于每个连接到远程集群的本地客户端来说，它们的地址表都是相同的，因此运行相同进程的客户端共享连接非常有用，这是通过共享HConnection实例来实现的。另外，当寻址失败时，连接有重试机制来刷新缓存，因此减少了客户端连接的初始化的开销。

另外HTablePool类也是受益的，所有连接池中的HTable实例都自动共用一个提供的Configuration实例，因此它们也共享连接。因为当客户想创建多个HTable实例时，最好先创建一个共用的Configuration实例：

Configuration conf = HBaseConfiguration.create();
HTable table_one = new HTable(conf , "table_one");
HTable table_two = new HTable(conf , "table_two");
...
HTable table_n = new HTable(conf , "table_n");

六，监控指标

1. RPC监控指标
2. JVM监控指标
3. INFO监控指标

RPC监控指标

master和region服务器还提供了RPC子系统的监控指标。它会自动监控每次不同客户端和服务器之间的操作，其中包括master和region服务器提供的RPC操作。

因为这些监控指标和客户端API直接相关，用户可以通过相应的API调用来推测它们的意思。虽然命名规则不完全一致。一个值得注意的模式是，与region相关的API操作有额外的region后缀，例如，HBaseAdmin提供的split()操作对应的splitRegion监控指标。只是少数监控指标和API没有关系如下

RPC Processing Time：服务器执行RPC消耗的时间，统计所有RPC调用的时间，并取平均值

RPC Queue Time：因为RPC需要使用排队系统，可能造成操作达到时间和操作实际执行时间的延迟，即排队时间

剩下的监控指标来自master和region服务器的RPC API，包括regionServerStartup()和regionServerReport，它们分别在region服务器初始化并向它的master节点报告时调用，以及常规报告状态时调用。

JVM监控指标

HBase 集群涉及的系统 HDFS，HBase，ZooKeeper都是用 Java 编写的，运行在 JVM 中，必须采集 JVM 相关的监控项。

Hbase 中对于 JVM 的监控数据，主要是 JvmMetrics 的对象来进行的。

JvmMetrics 主要统计的信息包括：内存的使用状态信息； GC 的统计信息；线程的统计信息；以及事件的统计信息。

INFO监控指标

HBase进程同样提供了一组称作info的监控指标。它们包括进程的固化信息，且用户可以自动检查这些值。这些监控指标只能通过JMX来获取。

HDFS引用HBase使用的 hadoop-core<X.Y-nnnn>.jar文件。这个JAR文件通常是安装包中已经提供的JAR文件，但是用户可以根据你的安装情况定制一个特定的JAR文件。返回的值如下所示：

这些值明显不能用图来显示，但是可以被管理员用来确认运行配置是否正确。

七，HBase和MapReduce集成

HBase最大特点之一就是可以紧密地与Hadoop的MapReduce框架集成。

框架

什么是MapReduce？

MapReduce被设计为在可扩展的方式下解决超过TB级数据处理过程中的问题。应当有一种方法可以建立一个性能随机数增加而线性提升的系统，这就是MapReduce努力要做到的。它遵守分治原则，通过将数据拆分到分布式文件系统中的不同机器上，让服务器（CPU或者新式的内核）能尽快的直接访问和处理数据。这种方法的问题是用户最终需要合并全局结果。同样,MpaReduce为解决这个问题将其内置了

1.切片

• 在FileInputFormat中，计算切片大小的逻辑：Math.max(minSize, Math.min(maxSize, blockSize))
• minSize的默认值是1,而maxSize的默认值是long类型的最大值,即可得切片的默认大小是blockSize(128M)
• maxSize参数如果调得比blocksize小，则会让切片变小，而且就等于配置的这个参数的值
• minSize参数调的比blockSize大，则可以让切片变得比blocksize还大
• hadoop为每个分片构建一个map任务,可以并行处理多个分片上的数据,整个数据的处理过程将得到很好的负载均衡,因为一台性能较强的计算机能处理更多的数据分片.
• 分片也不能切得太小,否则多个map和reduce间数据的传输时间,管理分片,构建多个map任务的时间将决定整个作业的执行时间.(大部分时间都不在计算上)

如果文件大小小于128M,则该文件不会被切片,不管文件多小都会是一个单独的切片,交给一个maptask处理.如果有大量的小文件,将导致产生大量的maptask,大大降低集群性能.

大量小文件的优化策略:

(1) 在数据处理的前端就将小文件整合成大文件，再上传到hdfs上，即避免了hdfs不适合存储小文件的缺点,又避免了后期使用mapreduce处理大量小文件的问题。(最提倡的做法)

(2)小文件已经存在hdfs上了，可以使用另一种inputformat来做切片(CombineFileInputFormat),它的切片逻辑和FileInputFormat（默认）不同，它可以将多个小文件在逻辑上规划到一个切片上，交给一个maptask处理。

2.环形缓存区

• 经过map函数的逻辑处理后的数据输出之后，会通过OutPutCollector收集器将数据收集到环形缓存区保存。
• 环形缓存区的大小默认为100M,当保存的数据达到80%时,就将缓存区的数据溢出到磁盘上保存。

3.溢出

• 环形缓存区的数据达到其容量的80%时就会溢出到磁盘上进行保存,在此过程中,程序会对数据进行分区(默认HashPartition)和排序(默认根据key进行快排)
• 缓存区不断溢出的数据形成多个小文件

4.合并

• 溢出的多个小文件各个区合并在一起(0区和0区合并成一个0区),形成大文件
• 通过归并排序保证区内的数据有序

5.shuffle

从过程2到过程7之间,即map任务和reduce任务之间的数据流称为shuffle(混洗),而过程5最能体现出混洗这一概念。一般情况下，一个reduce任务的输入数据来自与多个map任务，多个reduce任务的情况下就会出现如过程5所示的，每个reduce任务从map的输出数据中获取属于自己的那个分区的数据。

6.合并

运行reducetask的节点通过过程5，将来自多个map任务的属于自己的分区数据下载到本地磁盘工作目录。这多个分区文件通过归并排序合并成大文件，并根据key值分好组（key值相同的，value值会以迭代器的形式组在一起）。

7.reducetask

reducetask从本地工作目录获取已经分好组并且排好序的数据，将数据进行reduce函数中的逻辑处理。

8.输出

每个reducetask输出一个结果文件。

partition（分区）

数据从环形缓存区溢出到文件的过程中会根据用户自定义的partition函数进行分区，如果用户没有自定义该函数，程序会用默认的partitioner通过哈希函数来分区，hash partition 的好处是比较弹性，跟数据类型无关，实现简单，只需要设置reducetask的个数。分区的目的是将整个大数据块分成多个数据块，通过多个reducetask处理后，输出多个文件。通常在输出数据需要有所区分的情况下使用自定义分区，如在上述的流量统计的案例里，如果需要最后的输出数据再根据手机号码的省份分成几个文件来存储，则需要自定义partition函数，并在驱动程序里设置reduce任务数等于分区数（job.setNumReduceTasks(5);）和指明自己定义的partition（job.setPartitionerClass(ProvincePartitioner.class)）。在需要获取统一的输出结果的情况下，不需要自定义partition也不用设置reducetask的数量（默认1个）。

自定义的分区函数有时会导致数据倾斜的问题，即有的分区数据量极大，各个分区数据量不均匀，这会导致整个作业时间取决于处理时间最长的那个reduce，应尽量避免这种情况发生。

combiner（map端的reduce）

集群的带宽限制了mapreduce作业的数量，因此应该尽量避免map和reduce任务之间的数据传输。hadoop允许用户对map的输出数据进行处理，用户可自定义combiner函数（如同map函数和reduce函数一般），其逻辑一般和reduce函数一样，combiner的输入是map的输出，combiner的输出作为reduce的输入，很多情况下可以直接将reduce函数作为conbiner函数来使用（job.setCombinerClass(FlowCountReducer.class);）。combiner属于优化方案，所以无法确定combiner函数会调用多少次，可以在环形缓存区溢出文件时调用combiner函数，也可以在溢出的小文件合并成大文件时调用combiner。但要保证不管调用几次combiner函数都不会影响最终的结果，所以不是所有处理逻辑都可以使用combiner组件，有些逻辑如果在使用了combiner函数后会改变最后rerduce的输出结果（如求几个数的平均值，就不能先用combiner求一次各个map输出结果的平均值，再求这些平均值的平均值，这将导致结果错误）。

combiner的意义就是对每一个maptask的输出进行局部汇总，以减小网络传输量。（原先传给reduce的数据是（a，（1,1,1,1,1,1...））,使用combiner后传给reduce的数据变为（a，（4,2,3,5...）））

HBase MR类

1，InputFormat

第一个接触的类就是InputFormat，它负责两件事情：第一件事情是拆分输入数据，同时返回一个RcorderReader实例，这个实例定义了键值对象的类，并提供了next()方法来遍历输入的数据。

就HBase而言，它提供了一组专用的实现，叫做TableInputFormatBase，该实现的子类是TableInputFormat。TableInputFormatHBase实现了其中的大部分功能，单仍旧是抽象类，它的子类TableInputFormat是一个轻量级的实体类版本，同时也被很多我们提供的例子和真实的MapReduce类使用。

这些类提供了一个扫描HBase全表的实现。用户需要提供一个Scan实例；设置起止行键、添加过滤器和指定版本数目等。TableInputFormat将表拆分成大校合适的块，同时交给后面的MapReduce过程处理。

2，Mapper类

Mapper类构成了MapReduce过程的下一个阶段，同时也是其命名的原因之一，在这一阶段中，从RecorderReader读到的每一个数据都由一个map()方法来处理。Mapper读取一组特定的键/值对，但是可能会输出其他类型。这个过程非常便于将原始数据转换为更有用的数据类型，以做进一步处理。

HBase提供一个TableMapper类，将键的类型强制转换为一个ImmutableBytesWritable，同时将值的类型强制转换为Result类型，这些构成了TableRecordReader类返回的结果。

有一个TableMapper的特殊实现是IdentityTableMapper，它也是一个展示如何将自己想要的功能添加到HBase提供的类中比较好的例子。TableMapper类没有实现任何实际功能，它只是添加了键/值对的签名。IdentityTableMapper只是简单地把键/值对传到下一个处理阶段而已。

3，Reducer类

Reducer阶段和类的层次结构与Mapper阶段十分的相似，这次是得到了Mapper类的输出，同时这些输出会经历shuffler和sort处理。

在Mapper和Reducer阶段间的内部shuffle阶段中，每个不同的Map输出的中间结果都会被复制到Reducer服务器，同时sort阶段会将所有经过shuffled(copied)阶段处理的数据进行联合排序，这样Reducer得到的中间结果就是一个已经排序的数据集，在这个数据集中，每一个特定的键与它所有可能的值关联在一起。

4，OutputFormat

最后阶段由OutputFormat类处理，它的工作是将数据持久化到不同的位置，这里提供了一些具体的实现来允许结果能被输出到文件或者HBase表中。输出到HBase表时，用户可以使用TableOutputFormat类，它使用TableRecordWriter类将数据写到特定的HBase表中，

同时也要注意一下基数。一般情况下，有许多Mapper将记录传输给Reducer，但是只有一个OutputFormat处理它对应的Reducer输出。这是最后一个用于处理键/值对并将它们写到最终存储位置的类，这个存储位置可以是文件或者表。

Hadoop类需要TableOutputCommitter类来实现它们的功能。如果不需要HBase来存储数据，则它是一个空类，并没有实现实际的功能。不过其他OutputFormat的实现需要这个类。

当创建作业时候，输出表的名字就已经被指定了。除此之外，TableOutputFormat不会增加其他复杂性。其他值得注意的是，表的缓冲区自动刷写被关闭，同时缓冲区由系统内部自动管理。这对提高导入大量数据时的速度十分有帮助。

MAPReduce的执行地点

Hadoop的块备份策略对上层用户十分不透明；这个过程是自动完成的，用户不用为其操心。HBase依靠这个副本策略对上层提供持久化的存储。虽然这些对于上层完全透明，但是更进一步的问题是，这些会如何影响系统性能？

不论是基于Hadoop还是HBase，这个问题都是用户使用MR作业时会考虑的点。

Hadoop是怎么处理的呢？

MapReduce文档中指出任务在靠近其数据的地方执行（所谓近水楼台先得月，也不是没有道理的）。HDFS中存储都是按照块来的，默认是64MB或者128MB或者更大。每块数据都被指定到一个map任务处理。这也说明较大的块会减少所需map数目，因为map数目有需要被处理的块的数目决定的。Hadoop知道块的位置，所以就近原则在其最近的map上执行任务（实际块是3个备份的，系统会任意选择一个块数据来执行）。

HBase又是怎么处理的呢？

首先要考虑的就是HBase和Hadoop要不要在一个集群，如果是两个不同的集群怎么办？

答案是必须要在一个集群上。

假设如果HBase和Haddoop彼此之间是相互独立的，那么此时运行一个MR作业，作业却不能再相应的datanode上执行。如果任务所在的数据位置没有被提供给MR作业，将执行失败。所以HDFS和HBase的MR一定要在一个集群中才可。

所以其实也就很好理解了，HBase执行时候启动的MR作业要和HDFS的数据块严格意义上绑定在一起。

表拆分

当用户使用MapReduce作业从表中读取数据时，实质上是使用TableInputFormat取得数据。它符合MapReduce框架，并重载了公有的方法getSplit()和create RecordReader()。在一个作业被执行前，框架调用getSplit()来决定如何划分(chunk)，从而由块数目决定映射任务的数目。

对于HBase来说，TableInputFormat基于用户提供的Scan实例取得所需要的表的信息，简单地使用起止键来确定region，拆分块的数目与起止键之间的regin数目相等。如果用户没有设置这两个键，则所有的region都被包含在其中。

作业启动时，框架会按拆分的数目调用createRecorder()，并返回与当前块对应的TableRecordReader实例。换句话说，每个TableRecordReader实例处理一个对应的region，读取并遍历一个region的所有行。

每一个拆分也包含对应region所在的服务器的信息。正是这个信息能使HBase上的MapReduce作业本地化执行；框架会比较region对应的服务器名，如果有任务tracker在对应的服务器上运行，它会挑选这个服务器来执行作业。因为region服务器与Data Node运行在一个节点上，所以扫描会从本地磁盘读取文件。

挡在HBase上运行MapReduce作业时候，推荐用户关闭预测执行模式，这样会增加region和服务器负载，同时与本地运行相违背；预测作业在一个不同机器上运行，因此没有本地region服务器，这会增加网络带宽开销。

HBase读写分离

HBase有三个典型的API : read(get、scan)、write ，我们有时候希望这三个访问尽可能的互相不影响，如下配置：（线上默认没有配置读写分离）

场景

• 写请求与读请求都比较高，业务往往接受：写请求慢点可以，读请求越快越好，最好有单独的资源保障
• scan与get都比较多，业务希望scan不影响get（因为scan比较消耗资源）

相关配置：

• hbase.ipc.server.callqueue.read.ratio
• hbase.ipc.server.callqueue.scan.ratio

具体含义：

• hbase.ipc.server.callqueue.read.ratio 设置为0.5，代表有50%的线程数处理读请求
• 如果再设置hbase.ipc.server.callqueue.scan.ratio 设置为0.5，则代表在50%的读线程之中，再有50%的线程处理scan，也就是全部线程的25%

HBase还有基本特性都阅读完成，但是还是需要在实际任务中去通过解决问题来体会和加深HBase这些特性。