Amazon EMR HBase on S3 之二级索引、Thrift 和性能评测

HBase是作为Apache软件基金会Hadoop项目的一部分开发的开源、非关系、分布式数据库，为Hadoop生态系统提供非关系数据库功能。Amazon EMR从4.6.0版本开始，就提供了HBase。

Amazon EMR从5.2.0版本开始，就支持把HBase根目录和元数据直接存储到Amazon S3, 这样就实现了HBase的存算分离，使得数据变成了高可用。我们可以启动一个Amazon EMR集群，在使用HBase时将其目录指向S3中的HBase根目录位置。当关闭EMR集群后，HBase的数据文件仍然保留在S3上，如果启动新的EMR集群，HBase仍然可以使用原来位于S3的数据文件。关于HBase on S3, 请参考：https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-hbase-s3.html。

Amazon EMR 从5.7.0版本开始，HBase on S3支持只读副本集群。只读副本群集为只读操作提供对主集群数据文件和元数据的只读访问。这样就实现了HBase的读写分离。关于使用 HBase on S3 设置只读副本集群，请参考：https://aws.amazon.com/cn/blogs/china/setting-up-read-replica-clusters-with-hbase-on-amazon-s3/。

关于HBase on S3和HBase只读副本集群，上述文档里已经说的非常清楚了，这里不再赘述。本文从实战的角度，解释一下客户在选择使用HBase on S3的时候比较关心的两个问题，一个是如何把已有的HBase on HDFS迁移到HBase on S3，并包含二级索引的迁移；另一个是HBase on S3的性能问题。

1. HBase 的架构和文件

为了方便后面介绍HBase迁移的步骤，以及确保迁移过程中避免数据丢失，需要先了解HBase的架构和数据文件，来看存储在HDFS上的HBase的架构图：

图1: HBase on HDFS的架构和组成部分

下面依次介绍上图中的元素：

1. Client

提供了访问HBase的一系列API接口，如Java Native API、Rest API、Thrift API等，并维护Cache来加快对HBase的访问。

2. Zookeeper

HBase 通过 Zookeeper 来实现 Master 的高可用，保证任何时候集群中只有一个Master 、实时监控Region Server的上线和下线信息，并实时通知Master元数据的入口，以及集群配置等工作。

在创建Aamazon EMR时选中 HBase应用后，会自带创建一个Zookeeper应用。

3. HDFS

HDFS 为 HBase 提供底层数据存储服务。使用Amazon EMR HBase时，我们推荐使用S3替换HDFS.

4. Master（即图中的HMaster）

HBase通过Master来管理所有的 Region Server和对表的DDL操作.

5. Region Server（即图中的HRegionServer）

Region Server用来管理Region，处理外部对Region的IO请求，即对表的DML操作，向底层文件系统中读写数据。

Region Server管理多个Region，一个Region包含多个Store, 一个Store对应一个CF（列族），而一个Store包括位于内存中的Mem Store和位于磁盘的Store File（即HFile）。

6. Mem Store

写缓存，数据是先存储在 Mem Store 中，排好序后刷写 (flush) 到 Store File. 关于这个flush的触发机制，可以参考：https://www.jianshu.com/p/396664db17be .

7. Store File

实际的存储文件。Store File是以HFile的形式存储在HDFS（或者S3上）的。每个 Store 会有一个或多个 Store File.

8. HFile

文件格式，HBase的数据文件，即Store File是以HFile格式存储的。

默认情况下，HFile位于HDFS文件系统的/user/hbase/data目录下。如果是HBase on S3，则位于指定的S3目录。

9. HLog, 预写入日志，又称Write-Ahead Logs (WAL)

用来保存HBase的修改记录，当对HBase操作数据的时候，对数据的操作会先写在一个叫做Write-Ahead Log的文件中，然后再将操作的数据写入内存中。所以在系统出现故障的时候，可以通过这个日志文件来恢复数据。

默认情况下，HLog位于HDFS文件系统的/user/hbase/WALs目录下。

2. HBase 的二级索引

之所以会谈到二级索引，是因为我们在后面的HBase on HDFS迁移到HBase on S3时，也会涉及二级索引的迁移。这也是很多客户关心的话题。

HBase 的表数据按 RowKey 进行字典排序， RowKey 实际上是数据表的一级索引(Primary Index)，由于 HBase 本身没有二级索引 (Secondary Index) 机制，基于索引检索数据只能单纯地依靠 RowKey, 这使得HBase不能有效地支持多条件查询。

HBase 本身没不提供二级索引 (Secondary Index)，而是通过新建一个表的方式来实现实现二级索引的功能。为了实现索引而带来的额外的需求，例如更新数据时需要原子更新索引表，则需要在HBase上去开发实现，好在有开源的组件替HBase考虑到并实现了这些需求，例如 Apache Phoenix.

Phoenix提供了几种类型的二级索引，常用的是一种叫Covered Index（覆盖索引）的二级索引。这种索引在获取数据的过程中，内部不需要再去HBase表上获取任何数据，你查询需要返回的列的数据都被存储在索引中。要想达到这种效果，你的select 的列，where 的列，group by的列，都需要在索引中出现。

举个例子，如果你的SQL语句是：

SELECT "customer"."type" AS credit_card_type, count(*) AS num_customers 
FROM "customer" WHERE "customer"."state" = 'CA' GROUP BY "customer"."type";

要最大化查询效率和速度最快，你就需要建立覆盖索引：

CREATE INDEX my_index ON "customer" ("customer"."state") 
INCLUDE("PK", "customer"."city", "customer"."expire", "customer"."type");

我们可以参考文档： https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-phoenix-clients.html 来使用Phoenix建立HBase二级索引。在操作过程中，请注意：文档中还是使用sqlline-thin来执行Sql语句，新版本的Phoenix已经替换为psql了，例如：

/usr/lib/phoenix/bin/psql.py localhost:2181 /home/hadoop/phoenixQuery.sql

另外，如果遇到” a) Scanner id given is wrong, b) Scanner lease expired because of long wait between consecutive client checkins” 之类的错误提示，可以设置hbase.client.scanner.timeout.period为更大的值，例如3000000.

建立二级索引后，我们可以看到HBase里多了这样几张表，如下图：

图2: Phoenix 二级索引建立的HBase表 (customer表是HBase的测试表)

我们可以执行命令 ”/usr/lib/phoenix/bin/sqlline.py localhost:2181” 进入Phoenix执行环境，然后检查执行计划如下：

图3: 从执行计划看二级索引的效果

3. HBase on HDFS 迁移到 HBase on S3

HBase on S3的优点，很多文档和Blog已经介绍的很详细了，具体可以参考：https://docs.aws.amazon.com/zh_cn/emr/latest/ReleaseGuide/emr-hbase-s3.html . 我们总结起来就是：实现HBase的存算分离和读写分离，以及由此带来的各种优势，例如数据高可用、降本增效等。

需要指出的是：HBase on S3并不是指HBase所有的组成部分都放到S3，我们在章节1中提到的写缓存(Mem Store) 是内存部分，需要把写缓存的数据Push到位于S3的HFile文件里。另外HLog (WAL) 仍然是位于EMR Core节点的HDFS上。HBase on S3的组件分布图如下：

图4: HBase on S3的组件分布

由于很多客户已经在使用HBase on HDFS了，怎样帮助客户从HBase on HDFS迁移到 HBase on S3，就成了一个亟待解决的问题。在文章 https://aws.amazon.com/cn/blogs/china/tips-for-migrating-to-apache-hbase-on-amazon-s3-from-hdfs/ 中，提供了三种迁移的方法：

快照 (Snapshot)
导出和导入 (Export / Import)
CopyTable

上述三种方法，都是对单个表的迁移，在实际项目中可能多个表，包括二级索引产生的表，可能还需要自己写脚本实现整库的迁移。

还有一种迁移方式，就是把所有的库文件迁移到S3. 前提是把所有的表都Disable, 步骤如下：

1.在S3上创建HBase根目录，例如s3://dalei-demo/hbase

2.将HBase on HDFS集群的表Disable, 并刷新 ‘hbase:meta’

bash /usr/lib/hbase/bin/disable_all_tables.sh
hbase:001:0>flush 'hbase:meta'

3. 将HBase在HDFS上的文件全部Distcp到S3的HBase根目录

hadoop distcp hdfs://ip-10-0-0-126:8020/user/hbase/* s3://dalei-demo/hbase/

4.创建新的Amazon EMR集群，指定HBase on S3, 步骤可以参考：https://docs.aws.amazon.com/zh_cn/emr/latest/ReleaseGuide/emr-hbase-s3.html

5.在新的HBase on S3上Enable所有的表，然后做常规操作的测试，包括二级索引的测试

6.创建只读副本集群，并测试

到目前为止，HBase on HDFS 已经迁移到了HBase on S3, 有兴趣的读者可以再去测试一下数据文件的高可用性，例如把HBase on S3所在的Amazon EMR 集群终止，然后重建一个新的集群，配置HBase on S3指向老集群使用的S3上的HBase根目录，会发现表、索引、数据都可以正常使用。

请注意一点，上述测试可能因为数据没有Flush到HFile，而导致数据丢失，所以如果是生产环境，还是建议使用3个Master节点的Amazon EMR, 避免集群的突然崩溃。

如果是正常释放Amazon EMR集群，请一定执行迁移步骤中的第2）步，确保所有的数据和Meta Data，都被Flush到S3的HBase根目录下，避免可能出现的数据丢失。

4. 性能评测

实际开发中，客户是把HBase作为一种高并发、低延迟的No-Sql数据库来使用，并通过Thrift实现的接口来进行随机的读写。我们来对比一下HBase on HDFS和HBase on S3上的性能。

4.1. Thrift的编译和实现

Thrift是一种接口描述语言和二进制通讯协议，它被用来定义和创建跨语言的服务。它被当作一个远程过程调用 (RPC) 框架来使用，是由Facebook为“大规模跨语言服务开发”而开发的，它现在是Apache软件基金会的开源项目。

Thrift包含一套完整的栈来创建客户端和服务端程序。服务端的程序我们不需要关心，很多应用都自带了(Amazon EMR也带了Thrift Server). 客户端的接口是由Thrift提供的，客户端的代码是由Thrift根据这个接口文件生成的。生成的步骤请参考：https://thrift.apache.org/tutorial/py.html .

4.2. 通过Thrift访问HBase

Thrift为HBase提供了两个版本的服务端程序 (Thrift Server)：Thrift和Thrift2. Amazon EMR HBase默认启动的Thrift，目前大部分客户都在使用Thrift2, 我们先把启动的Thrift Server的版本改一下，步骤如下：

1.关闭 Thrift 服务

sudo systemctl stop hbase-thrift

2.修改 hbase-thrift 服务的启动脚本

sudo vim /etc/systemd/system/hbase-thrift.service

将 ExecStart=/usr/lib/hbase/bin/hbase-daemon.sh start thrift
替换为 ExecStart=/usr/lib/hbase/bin/hbase-daemon.sh start thrift2

将 /var/run/hbase/hbase-hbase-thrift.pid
替换为 /var/run/hbase/hbase-hbase-thrift2.pid

3. 刷新 systemd 服务配置

sudo systemctl daemon-reload

4.重新启动 hbase-thrift 服务，此时启动的是 Thrift2

sudo systemctl start hbase-thrift

5.检查是否启动成功

ps aux | grep thrift2

通过Thrift2访问HBase的步骤如下：

1.安装Python依赖包

pip install thrift
pip install hbase-thrift

2.部署客户端代码

将上一节编译生成的Thrift客户端代码中的ttypes.py和THBaseService.py, 放到Python依赖包生成的目录，例如 /home/hadoop/.local/lib/python3.7/site-packages/hbase . 如果已有ttypes.py文件，则替换它。

如果自己编译失败，也可以从 https://github.com/xudalei1977/hbase-thrift-performance 直接下载这两个文件，放到上面的目录里。

3.在hbase shell里创建表空间和表

hbase:001:0> create_namespace 'test_ns'
Took 8.4409 seconds                                                                                                                                                                                     
hbase:002:0> create 'test_ns:test_1', {NAME =>'cf_1', COMPRESSION => 'snappy', TTL=>'86400' }, { NUMREGIONS => 257, SPLITALGO => 'HexStringSplit' }
Created table test_ns:test_1
Took 48.2039 seconds                                                                                                                                                                                    
=> Hbase::Table - test_ns:test_1

4.执行的测试代码如下：

from thrift.transport import TSocket
from thrift.protocol import TBinaryProtocol
from thrift.transport import TTransport
from hbase.ttypes import *
from hbase import THBaseService

transport = TTransport.TBufferedTransport(TSocket.TSocket('127.0.0.1', 9090))
protocol = TBinaryProtocol.TBinaryProtocolAccelerated(transport)
client = THBaseService.Client(protocol)
transport.open()

table = 'test_ns:test_1'
row = 'row1'

put_columns = [
        TColumnValue('cf_1'.encode(), 'col_1'.encode(), 'value_1'.encode()),
        TColumnValue('cf_1'.encode(), 'col_2'.encode(), 'value_2'.encode()),
        TColumnValue('cf_1'.encode(), 'col_3'.encode(), 'value_3'.encode()),
        TColumnValue('cf_1'.encode(), 'col_4'.encode(), 'value_4'.encode()),
        TColumnValue('cf_1'.encode(), 'col_5'.encode(), 'value_5'.encode())
        ]
tput = TPut(row.encode(), put_columns)
client.put(table.encode(), tput)

get_columns = [
        TColumn('cf_1'.encode(), 'col_1'.encode()),
        TColumn('cf_1'.encode(), 'col_2'.encode()),
    ]
tget = TGet(row.encode(), get_columns)
tresult = client.get(table.encode(), tget)
print(tresult)

transport.close()

4.3. 性能对比

我们会在HBase on HDFS和HBase on S3上分别测试写入 (Put) 操作和读取 (Get) 操作的性能。首先创建两个Amazon EMR集群，一个使用HBase on HDFS, 一个使用HBase on S3, 版本采用最新的6.9.0, 配置都是如下：

角色	实例类型	vCPU	内存(GiB)	磁盘(GiB)	数量
Master	m6g.8xlarge	32	122	256	1
Core	m6g.4xlarge	16	61	128	5

测试代码位于 https://github.com/xudalei1977/hbase-thrift-performance . 用户可以git clone到Amazon EMR的Master节点上。

我们会用到Parallel模拟并发测试，先下载Parallel源文件到Master节点，并进行编译如下：

wget https://ftpmirror.gnu.org/parallel/parallel-20221122.tar.bz2
tar -jxvf parallel-20221122.tar.bz2 
cd parallel-20221122 
./configure 
make && sudo make install

测试写入 (Put) 操作的性能的代码是hbase-put.py, 我们使用如下Shell执行：

CONF_FILE=parallel.hbase
rm -rf $CONF_FILE
CORE_NUM=`nproc`
KEY_NUM=4
TASK_NUM=`expr $CORE_NUM \* $KEY_NUM`

for i in $(seq 1 $TASK_NUM);do echo "/usr/bin/python3 ~/hbase-put.py 400000 1" >> $CONF_FILE ; done;

wc -l $CONF_FILE

nohup parallel -j $TASK_NUM < $CONF_FILE &

在上面的代码中，请注意：

Parallel根据当前节点的核数，乘以每个节点的Task数目，来确定执行hbase-put.py文件的并发数
hbase-put.py用来写入数据到表 ”test_ns:test_1”, 后面的参数表示每次执行写入的条数和线程数，由于Parallel已经使用了并发，这里线程数就指定为1，可以修改这两个参数来调整写入的记录条数
在写入数据时，RowKey是采用了md5作用于随机数，保证了数据在Region上的均匀分布

写入 (Put) 操作的性能对比如下：

图5: HBase on HDFS 和HBase on S3的写入 (Put) 操作的性能对比

可以看出，二者在写入时的性能，相差无几。
再来看读取 (Get) 操作，性能测试的代码是hbase-get.py, 我们使用如下Shell执行：

CONF_FILE=parallel.hbase
rm -rf $CONF_FILE
CORE_NUM=`nproc`
KEY_NUM=4
TASK_NUM=`expr $CORE_NUM \* $KEY_NUM`

for i in $(seq 1 $TASK_NUM);do echo "/usr/bin/python3 ~/hbase-get.py 400000 1" >> $CONF_FILE ; done;

wc -l $CONF_FILE

nohup parallel -j $TASK_NUM < $CONF_FILE &

在上面的代码中，请注意：

Parallel根据当前节点的核数，乘以每个节点的Task数目，来确定执行hbase-get.py文件的并发数
hbase-get.py用来从表 ”test_ns:test_1”中读取数据, 后面的参数表示每次执行读取的次数和线程数，由于Parallel已经使用了并发，这里线程数就指定为1，可以修改这两个参数来调整读取的次数
在读取数据时，采用了md5作用于随机数去匹配Rowkey, 有可能因为匹配不到Rowkey而遍历所有的HFile，这保证了读取的数据不是只位于Mem Store, 也有位于HDFS或者S3上的HFile里的数据

读取 (Get) 操作的性能对比如下：

可以看出在进行读操作时，HBase on S3有性能优势。

关于其它的操作，不论是HBase自带的Count, Scan, Filter，还是Spark读写HBase，还是我们前面介绍的二级索引，作者在实际开发中都做过比较，HBase on HDFS和HBase on S3的性能差别不大。此外，如果使用只读集群实现HBase的读写分离的话，还可获得更多的优化空间，因为读和写可以设置不同的参数。有兴趣的朋友可以自己测试一下。

总之，还是鼓励HBase的用户，把数据迁到S3上来。