使用 Athena 替换 Hbase 实现对历史数据的查询分析

阅读数:15 2019 年 11 月 27 日 08:00

使用 Athena 替换 Hbase 实现对历史数据的查询分析

Apache HBase 是一种构建在 HDFS 之上的分布式、面向列的存储系统。在需要实时读写、随机访问超大规模数据集时,作为用户的首选分布式数据库;但 HBase 也有其局限性,譬如说不支持 SQL 语句查询,随着数据规模的变大,造成用户成本的大幅增加,其稳定性和故障恢复能力会变差,也给运维人员带来很大的挑战;我们在拜访客户时也发现有些用户将大量的历史数据和在线数据全部存储在 HBase 中也遇到了上述问题。因而本文将介绍一种帮助用户从 HBase 数据库中剥离出历史数据,减小 HBase 数据库的规模,提高其稳定性并大幅降低客户的成本,实现对历史数据的查询的方案。

Apache HBase 特点及应用场景

Apache HBase 是一个分布式,版本化,面向列的开源数据库,构建在 Aapche Hadoop 和 Aapche ZooKeeper 之上。它特别适合千万级的高并发海量数据的瞬间写入,而相对读数据量小的应用,支持存储结构化和非结构化数据和数据的多版本化。但它不适合大范围的扫描查询和支持多条件的查询,不支持基于 SQL 语句的查询。

Amazon Athena 特点及应用场景

Amazon Athena 是一种交互式查询服务,让您能够轻松使用标准 SQL 直接分析 Amazon S3 中的数据。只需在 AWS 管理控制台中单击几下,客户即可将 Athena 指向自己在 S3 中存储的数据,然后开始使用标准 SQL 执行临时查询并在数秒内获取结果。Athena 是 Serverless 服务,因此没有需要设置或管理的基础设施,客户只需为其执行的查询付费。它特别适合使用 Athena 处理日志、执行即席分析以及运行交互式查询。Athena 可以自动扩展并执行并行查询,因此可快速获取结果,对于大型数据集和复杂查询也不例外。

本次实验演示的 Demo 数据来自于纽约出租车公司公布的公开数据源 2017 年 1 月到 6 月份 Green Taxi trip 数据,下载链接: https://www1.nyc.gov/site/tlc/about/tlc-trip-record-data.page。

准备模拟数据并上传到 S3 Bucket

登陆 AWS 管理控制台,创建 S3 bucket 命名为 nytaxisdata,然后创建三个文件夹分别命名为 green,hbasedata,hbaseexport 如下图所示:

使用 Athena 替换 Hbase 实现对历史数据的查询分析

处理数据

清理掉原始数据中的空格列,运行如下命令:

Bash

复制代码
awk 'BEGIN{FS=OFS=","}{gsub(/ /,"-",$2);gsub(/ /,"-",$3);print $0}' green_tripdata_2017-01.csv |more
[hadoop@ip-172-31-28-170 ~]$ awk 'BEGIN{FS=OFS=","}{gsub(/ /,"-",$2);gsub(/ /,"-",$3);print $2,$3,$1,$6,$7}' green_tripdata_2017-01.csv >>greentrip01.csv

如下图所示:

使用 Athena 替换 Hbase 实现对历史数据的查询分析

分别依次处理完对应的 6 个 excel 文件。然后将处理完的 2017 年 1 月到 6 月份的数据文件上传到 green 文件夹内,如下图所示:

使用 Athena 替换 Hbase 实现对历史数据的查询分析

创建 EMR 集群

登陆到 AWS 管理控制台,选择 EMR 服务,点击创建集群,如下图所示:

使用 Athena 替换 Hbase 实现对历史数据的查询分析

点击 Go to Advanced options,选择 Hadoop 和 HBase 服务如下图所示:

使用 Athena 替换 Hbase 实现对历史数据的查询分析

在 Storage Mode 中选择 S3,并设置上面刚创建的 bucket 路径 s3://nytaxisdata/hbasedata, 如下图所示:

使用 Athena 替换 Hbase 实现对历史数据的查询分析

在 Hardware Configuration 设置中选择 Uniform instance groups,并设置网络,子网组及根卷大小,此处设置为 100G,如下图所示:

使用 Athena 替换 Hbase 实现对历史数据的查询分析

点击下一步,设置 EMR 集群的节点类型和实例数,如下图所示:

使用 Athena 替换 Hbase 实现对历史数据的查询分析

点击下一步输入集群名称 hbasecluster,其他默认即可,如下图所示:

使用 Athena 替换 Hbase 实现对历史数据的查询分析

点击 Next,设置 EC2 key pair,如下图所示:

使用 Athena 替换 Hbase 实现对历史数据的查询分析

点击 Create cluster,等待集群创建完成。

登陆 HBase 集群

创建一个表名为 taxiinfo 的表,指定列簇为 userinfo 和 Others,运行如下命令:

Bash

复制代码
hbase(main):001:0> create 'taxiinfo','dropofftime','comno','others'

如下图所示:

使用 Athena 替换 Hbase 实现对历史数据的查询分析

导入数据到 HBase 集群

运行如下命令:

Bash

复制代码
sudo hbase org.apache.hadoop.hbase.mapreduce.ImportTsv -Dimporttsv.separator="," -Dimporttsv.columns=HBASE_ROW_KEY,dropofftime,comno,others:distance,others:length taxiinfo s3://nytaxisdata/green/greentrip01.csv

一次运行上面的命令将 greentrip01.csc 到 greentrip06.csv 文件全部导入到 hbase 数据库名为 taxiinfo 中。

统计表中的数据总行数,运行命令 count,如下所示:

Bash

复制代码
hbase(main):002:0> count 'taxiinfo',INTERVAL=>100000
Current count: 100000, row: 2017-01-04-22:58:55
Current count: 200000, row: 2017-01-08-16:39:14
Current count: 300000, row: 2017-01-12-15:35:40
……
Current count: 4300000, row: 2017-06-04-03:39:03
Current count: 4400000, row: 2017-06-08-09:56:53
Current count: 4500000, row: 2017-06-11-18:32:27
Current count: 4600000, row: 2017-06-15-20:14:35
Current count: 4700000, row: 2017-06-19-13:37:54
Current count: 4800000, row: 2017-06-23-15:21:13
Current count: 4900000, row: 2017-06-27-14:46:20
4989642 row(s) in 257.1330 seconds
=> 4989642

至此,验证 HBase 中的数据准备完毕。

剥离 HBase 数据库中的历史数据

根据需要用户可以将 HBase 数据库中的历史数据进行导出,HBase 数据库中的数据导出为 CSV 文件有多种方式,譬如自己编写 MapReduce 类实现,或者借助 Pig 应用等,本文采用 happybase 包实现。HappyBase 是方便开发人员通过 python 实现与 HBase 进行交互的开发库,通过编写 Python 脚本方式非常灵活的将任意条件的存储在 HBase 中的数据抽取处理转换成 CSV 文件。

安装 HappyBase

登陆到 EMR Master 节点,运行如下命令:

Bash

复制代码
[hadoop@ip-172-31-36-88 ~]$ sudo pip install happybase

如下图所示:

使用 Athena 替换 Hbase 实现对历史数据的查询分析

导出 HBase 数据

编写 Python 脚本如下所示:

Python

复制代码
import happybase, sys, os, string
# 设置导出数据目录 /mnt
# Output directory for CSV files
outputDir = "/mnt"
# HBase Thrift server to connect to. Leave blank for localhost
server = ""
# 连接到 HBASE 数据库
c = happybase.Connection(server)
# Get the full list of tables
tables = c.tables()
# 轮询数据库中的所有表
for table in tables:
# 写文件
file = open(outputDir + "/" + table + ".csv", "w")
t = c.table(table)
print table + ": ",
count = 0
for prefix in string.printable:
try:
for key, data in t.scan(row_prefix=prefix):
# 第一个 key
if count == 0:
startRow = key
for col in data:
value = data[col]
column = col[col.index(":")+1:]
# 写行,列,值
file.write("%s, %s, %s\n" % (key, column, value))
count += 1
except:
os.system("hbase-daemon.sh restart thrift")
c = happybase.Connection(server)
t = c.table(table)
continue
# 最后一个 key
endRow = key
print "%s => %s, " % (startRow, endRow),
print str(count)

保存运行命令:

Bash

复制代码
python hbase_export_csv.py

如下图所示数据导出 4989642:

使用 Athena 替换 Hbase 实现对历史数据的查询分析

将导出的 csv 文件上传到 s3://nytaxisdata/exportdata 文件夹下,如下图所示:

使用 Athena 替换 Hbase 实现对历史数据的查询分析

使用 Amazon Athena 实现对历史数据的查询

打开 AWS Athena 管理控制台,运行如下命令:

Bash

复制代码
CREATE EXTERNAL TABLE `hbaseexport`(
`col0` string,
`col1` string,
`col2` string)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
STORED AS INPUTFORMAT
'org.apache.hadoop.mapred.TextInputFormat'
OUTPUTFORMAT
'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION
's3://nytaxisdata/hbaseexport/'

如下图所示:

使用 Athena 替换 Hbase 实现对历史数据的查询分析

数据查询

在 SQL 脚本编写区,运行如下查询:

Bash

复制代码
SELECT * FROM "nymetadatastore"."hbaseexport" limit 10;

如下图所示:

使用 Athena 替换 Hbase 实现对历史数据的查询分析

总结

通过本篇您将了解到如何使用使用 HappyBase 库编写 Python 脚本实现将 HBase 数据库中的历史数据的随意抽取并转换成 csv 文件,利用 Athena 实现对存储在 S3 上数据文件的访问。该方案可以帮助客户减小 HBase 集群,大幅降低数据的访问成本。

作者介绍:

!
复制代码
王友升拥有超过 13 年的 IT 从业经验,负责基于 AWS 的云计算方案架构咨询和设计,推广 AWS 云平台技术和各种解决方案。在加入 AWS 之前,王友升曾在中地数码,浪潮,惠普等公司担任软件开发工程师、DBA 和解决方案架构师。他在服务器、存储、数据库优化方面拥有多年的经验,同时对大数据、Openstack 及人工智能和机器学习方面也进行一定的研究和积累。

本文转载自 AWS 技术博客。

原文链接:
https://amazonaws-china.com/cn/blogs/china/using-athena-to-replace-hbase-to-query-analyze-historical-data/

欲了解 AWS 的更多信息,请访问【AWS 技术专区】

评论

发布