数据导入HBase最常用的三种方式及实践分析(2)

HBase提供importtsv工具支持从TSV文件中将数据导入HBase。使用该工具将文本数据加载至HBase十分高效，因为它是通过MapReduce Job来实施导入的。哪怕是要从现有的关系型数据库中加载数据，也可以先将数据导入文本文件中，然后使用importtsv 工具导入HBase。在导入海量数据时，这个方式运行的很好，因为导出数据比在关系型数据库中执行SQL快很多。
importtsv 工具不仅支持将数据直接加载进HBase的表中，还支持直接生成HBase自有格式文件（HFile），所以你可以用HBase的bulk load工具将生成好的文件直接加载进运行中的HBase集群。这样就减少了在数据迁移过程中，数据传输与HBase加载时产生的网络流量。下文描述了importtsv 和bulk load工具的使用场景。我们首先展示使用importtsv 工具从TSV文件中将数据加载至HBase表中。当然也会包含如何直接生成HBase自有格式文件，以及如何直接将已经生成好的文件加载入HBase。

准备
我们在本文中将使用 “美国国家海洋和大气管理局 气候平均值”的公共数据集合。访问http://www1.ncdc.noaa.gov/pub/data/normals/1981-2010/下载。 我们使用在目录 products | hourly 下的小时温度数据（可以在上述链接页面中找到）。下载hly-temp-10pctl.txt文件。
下载后的数据因为格式不支持的原因，不能直接用importtsv工具加载。我们提供了脚本来帮助你将数据转换为TSV文件。除了原有数据，被加载的TSV文件中还必须包含一个栏位用于表示HBase表数据行的row key。本文附带的_tsv_hly.py脚本从NOAA的小时数据文件中读取数据，生成row key并将数据输出至本地文件系统的TSV文件:
1

$ python to_tsv_hly.py -f hly-temp-10pctl.txt -t hly-temp-10pctl.tsv

因为importtsv工具是通过运行MapReduce Job来实施导入动作，我们需要在集群上运行MapReduce。在主节点上执行下述命令以开启MapReduce守护进程：
1

hadoop$ $HADOOP_HOME/bin/start-mapred.sh

我们在客户端服务器上添加hac用户用于运行job；建议在生产环境如此实施。为了能够从客户端运行MapReduce Job，你需要将${hadoop.tmp.dir}目录的写权限开放给客户端的hac用户。我们假设${hadoop.tmp.dir}目录为/usr/local/hadoop/var:
1
2

root@client1# usermod -a -G hadoop hac
root@client1# chmod -R 775 /usr/local/hadoop/var

在HDFS中为hac用户建立主文件夹：
1
2

hadoop@client1$ $HADOOP_HOME/bin/hadoop fs -mkdir /user/hac
hadoop@client1$ $HADOOP_HOME/bin/hadoop fs -chown hac /user/hac

同时也确认hac用户在HDFS中的MapReduce的临时目录中也有写权限：
1

hadoop@client1$ $HADOOP_HOME/bin/hadoop fs -chmod -R 775 /usr/local/hadoop/var/mapred

如何实施
使用MapReduce将数据从TSV文件加载至HBase的table，按照如下步骤实施：
1.在HDFS中建立文件夹，并且将TSV文件从本地文件系统拷贝至HDFS中：
1
2

hac@client1$ $HADOOP_HOME/bin/hadoop fs -mkdir /user/hac/input/2-1
hac@client1$ $HADOOP_HOME/bin/hadoop fs -copyFromLocal hly-temp-10pctl.tsv /user/hac/input/2-1

2.在HBase中添加目标表。连接到HBase，添加hly_temp表：
1
2

hac@client1$ $HBASE_HOME/bin/hbase shell
hbase> create 'hly_temp', {NAME => 't', VERSIONS => 1}

3.若表已经存在（上一节中已经建好），则添加一个新列族：
1
2
3

hbase> disable 'hly_temp'
hbase> alter 'hly_temp', {NAME => 't', VERSIONS => 1}
hbase> enable 'hly_temp'

4.将hbase-site.xml文件放置在Hadoop的配置目录中就能够加入Hadoop的环境变量了：
1

hac@client1$ ln -s $HBASE_HOME/conf/hbase-site.xml $HADOOP_HOME/conf/hbase-site.xml

5.编辑客户端服务器的$HADOOP_HOME/conf 下的hadoop-env.sh文件，添加HBase的依赖库到Hadoop的环境变量中：
1
2

hadoop@client1$ vi $HADOOP_HOME/conf/hadoop-env.sh
export HADOOP_CLASSPATH=/usr/local/zookeeper/current/zookeeper-3.4.3.jar:/usr/local/hbase/current/lib/guava-r09.jar

6.使用hac用户运行importtsv工具，执行如下脚本：
1
2
3
4
5

hac@client1$ $HADOOP_HOME/bin/hadoop jar $HBASE_HOME/hbase-
0.92.1.jar importtsv \ -Dimporttsv.columns=HBASE_ROW_KEY,t:v01,t:v02,t:v03,t:v04,t:v05,t:v06,t:v07,t:v08,t:v09,t:v10,t:v11,t:v12,t:v13,t:v14,t:v15,t:v1
6,t:v17,t:v18,t:v19,t:v20,t:v21,t:v22,t:v23,t:v24 \
hly_temp \
/user/hac/input/2-1

7.通过MapReduce Job管理页面—http://master1:50030/jobtracker.jsp检查任务状态。
8. 验证HBase目标表中的导入数据。通过验证hly_temp表中的数据总量，并且还要检查表中的一些样本数据。表中的数据总量应该于文件中的行数相同。表中的row key应该与文件中的第一个字段相同。每行数据都有t:v01, t:v02, …, t:v24等单元格，每个单元格的值都应当与TSV文件中的栏位相同：
1
2
3
4
5
6
7
8

hbase> count 'hly_temp'
95630 row(s) in 12.2020 seconds
hbase> scan 'hly_temp', {COLUMNS => 't:', LIMIT => 10}
AQW000617050110 column=t:v23,
timestamp=1322959962261, value=781S
AQW000617050110 column=t:v24,
timestamp=1322959962261, value=774C
10 row(s) in 0.1850 seconds

运行原理
Importtsv工具只从HDFS中读取数据，所以一开始我们需要将TSV文件从本地文件系统拷贝到HDFS中，使用的是hadoop fs -copyFromLocal命令。在步骤2中，我们在HBase中建立了表(hly_temp) 以及列族 (t)。若表已经存在，我们可以修改表，加入列族。所有的数据都加载进新建的列族中，已经存在原有列族中的数据则不受影响。运行MapReduce Job，需要使用hadoop的jar命令来运行包含class编译文件的JAR文件。为了在命令行中能够使用HBase的配置信息，我们将hbase-site.xml放到$HADOOP_HOME/conf 目录下，从而产生关联；在该目录下的所有文件都会被hadoop命令行加入到Java进程的环境变量中。
步骤5中，设置hadoop-env.sh中的HADOOP_CLASSPATH以加入运行时依赖。除了ZooKeeper库之外，guava-r09.jar也是importtsv运行依赖库，它是用于转换TSV文件的库。
Importtsv本身是一个在HBase的JAR文件中的JAVA类。在步骤6中，我们通过hadoop的jar命令来运行该工具。这个命令会启动一个Java进程，并且自动添加所有的依赖。需要运行哪个JAR是通过指定hadoop jar命令的第一个参数，在这里是使用hbase-0.92.1.jar。
下列参数要被传递至hbase-0.92.1.jar的主类：

     TSV文件的字段索引与HBase表中列的对应信息是对 -Dimporttsv.columns参数进行设置，在本文中，TSV文件格式是(rowkey, value1, value2, …, value24)。我们将数据存入HBase的列族 t 中，使用v01 对应 value1, v02 对应value2等类似方式。HBASE_ROW_KEY 中存放的就是row key字段。
     在 -Dimporttsv.columns 参数之后，我们还需要为命令行指定表名参数（hly_temp）以及TSV文件路径 (/user/hac/input/2-1)参数

还有一些其他选项可以被指定。运行importtsv不带任何参数就会打印出使用信息摘要：
1
2
3
4
5

hac@client1$ $HADOOP_HOME/bin/hadoop jar $HBASE_HOME/hbase-0.92.1.jar
importtsv
用法: importtsv -Dimporttsv.columns=a,b,c
将指定路径的TSV数据导入指定的表中。
…

其他包含 -D的可指定的选项包括：
-Dimporttsv.skip.bad.lines=false – 若遇到无效行则失败
‘-Dimporttsv.separator=|’ – 文件中代替tabs的分隔符
-Dimporttsv.timestamp=currentTimeAsLong – 导入时使用指定的时间戳
-Dimporttsv.mapper.class=my.Mapper – 使用用户指定的Mapper类来代替默认的org.apache.hadoop.hbase.mapreduce.TsvImporterMapper
这个工具启动了MapReduce Job。在job的map阶段，它从指定路径的TSV文件中读取并转换，然后根据栏位映射信息将数据写入HBase的table中。此处读和写的操作是在多台服务器上并行执行，所以相比从单台节点读取速度快很多。该job中默认是没有reduce阶段。我们能够在MapReduce的管理页面上查看job的进度，统计以及其他MapReduce信息。
需要查看表中插入的数据，可以使用HBase Shell中的scan命令。我们可以指定列为 ‘t’（COLUMNS => ‘t:’）来只对表中的t列族进行查询。

更多内容
Importtsv工具默认使用了HBase的Put API来将数据插入HBase表中，在map阶段使用的是TableOutputFormat 。但是当 -Dimporttsv.bulk.输入选项被指定时，会使用HFileOutputFormat来代替在HDFS中生成HBase的自有格式文件（HFile）。而后我们能够使用completebulkload 来加载生成的文件到一个运行的集群中。根据下列步骤可以使用bulk 输出以及加载工具：
1.在HDFS中建立文件夹用于存放生成的文件：
1

hac@client1$ $HADOOP_HOME/bin/hadoop fs -mkdir /user/hac/output

2.运行importtsv并加上bulk输出选项：
1
2
3
4
5
6

hac@client1$ $HADOOP_HOME/bin/hadoop jar $HBASE_HOME/hbase-
0.92.1.jar importtsv \
-Dimporttsv.bulk.output=/user/hac/output/2-1 \
-Dimporttsv.columns=HBASE_ROW_KEY,t:v01,t:v02,t:v03,t:v04,t:v05,t:v06,t:v07,t:v08,t:v09,t:v10,t:v11,t:v12,t:v13,t:v14,t:v15,t:v16,t:v17,t:v18,t:v19,t:v20,t:v21,t:v22,t:v23,t:v24 \
hly_temp \
/user/hac/input/2-1

3.完成bulk加载：
1
2
3

hac@client1$ $HADOOP_HOME/bin/hadoop jar $HBASE_HOME/hbase-0.92.1.jar completebulkload \
/user/hac/output/2-1 \
hly_temp

completebulkload 工具读取生成的文件，判断它们归属的族群，然后访问适当的族群服务器。族群服务器会将HFile文件转移进自身存储目录中，并且为客户端建立在线数据。