1. 前言
MapReduce是一种经典的分布式计算模型，由Google提出并被Hadoop等开源项目广泛应用。HBase是一个分布式的非关系型数据库系统，它使用Hadoop作为底层存储。在实际应用中，我们可能需要将HBase中的数据复制到其他存储系统中，比如MySQL、Hive等。本文介绍如何使用通用的MapReduce程序复制HBase表数据。
2. 准备工作
2.1 环境准备
在开始之前，我们需要保证有一个Hadoop集群和一个HBase集群，以及相应的依赖包。我们还需要在本地开发环境中安装相关工具，如Maven等。
2.2 HBase表结构
在进行HBase表数据的复制之前，我们需要先创建一个HBase表。这里假设我们要复制的表名为source_table，表中包含三个列族info1, info2, info3，并且rowkey的长度为10字节，info1:column1和info2:column2的值的类型为字符串，info3:column3的值的类型为整数。创建表的命令如下：
create 'source_table', {NAME => 'info1', VERSIONS => 1}, {NAME => 'info2', VERSIONS => 1}, {NAME => 'info3', VERSIONS => 1}
2.3 复制目标表结构
我们还需要在目标数据库中创建一个表，与源表结构相同。
3. MapReduce程序编写
3.1 代码结构
我们通过编写一个MapReduce程序，将HBase表数据复制到其他存储系统中。代码目录结构如下：
.
├── pom.xml
├── src
│ ├── main
│ │ ├── java
│ │ │ └── com
│ │ │ └── example
│ │ │ └── CopyHBaseData.java
│ │ └── resources
│ └── test
│ └── java
└── target
├── classes
└── ...
其中，pom.xml是Maven项目的配置文件，src/main/java/com/example/CopyHBaseData.java是MapReduce程序的核心代码，实现了从源表中读取数据，并将数据写入到目标表中。
3.2 代码实现
主要分为三个部分：Job配置、Mapper实现和Reducer实现。
3.2.1 Job配置
首先需要在程序中定义一个Job，并设置相关的参数，如数据输入路径、Mapper和Reducer的Class等。代码如下：
Configuration conf = HBaseConfiguration.create();
Job job = Job.getInstance(conf, "CopyHBaseData");
job.setJarByClass(CopyHBaseData.class);
job.setMapperClass(CopyHBaseDataMapper.class);
job.setReducerClass(CopyHBaseDataReducer.class);
job.setOutputKeyClass(ImmutableBytesWritable.class);
job.setOutputValueClass(Put.class);
FileInputFormat.addInputTable(job, new TableName("source_table"));
FileOutputFormat.setOutputPath(job, new Path("/"));
conf是配置信息，将会与默认配置覆盖，使用默认配置信息。Job名称为CopyHBaseData，相关程序为本程序的类。同时设置Mapper和Reducer的实现类为CopyHBaseDataMapper和CopyHBaseDataReducer。数据输入路径设置为HBase表名
3.2.2 Mapper实现
该部分实现从源HBase表中读取数据。
public static class CopyHBaseDataMapper
extends TableMapper<ImmutableBytesWritable, Put> {
private ImmutableBytesWritable outkey = new ImmutableBytesWritable();
@Override
public void map(ImmutableBytesWritable key, Result value, Context context)
throws IOException, InterruptedException {
Put put = new Put(key.copyBytes());
// info1:column1
byte[] a = value.getValue(Bytes.toBytes("info1"), Bytes.toBytes("column1"));
if (a != null) {
put.addColumn(Bytes.toBytes("info1"), Bytes.toBytes("column1"), a);
}
// info2:column2
byte[] b = value.getValue(Bytes.toBytes("info2"), Bytes.toBytes("column2"));
if (b != null) {
put.addColumn(Bytes.toBytes("info2"), Bytes.toBytes("column2"), b);
}
// info3:column3
byte[] c = value.getValue(Bytes.toBytes("info3"), Bytes.toBytes("column3"));
if (c != null) {
put.addColumn(Bytes.toBytes("info3"), Bytes.toBytes("column3"), c);
}
outkey.set(Bytes.toBytes(key.toString()));
context.write(outkey, put);
}
}
首先我们实现了该类，并继承自HBase中的TableMapper。使用ImmutableByteWritable来表示行键，Put来表示修改数据的类型。Map阶段会将输入Key是一个行键，而Value则是一个包含该行键的所有列和值的Result对象。我们逐一按照列族和列名获取数据并放入Put，在结果输出的时候，用ImmutableByteWritable作为key，Put作为value输出，以备Reduce使用。
3.2.3 Reducer实现
该部分实现将数据写入到目标表中。
public static class CopyHBaseDataReducer
extends TableReducer<ImmutableBytesWritable, Put, ImmutableBytesWritable> {
@Override
public void reduce(ImmutableBytesWritable key, Iterable<Put> values, Context context)
throws IOException, InterruptedException {
Configuration conf = context.getConfiguration();
Connection connection = ConnectionFactory.createConnection(conf);
Table table = connection.getTable(TableName.valueOf("target_table"));
for (Put put : values) {
table.put(put);
}
table.close();
connection.close();
}
}
Reducer的输出Key也是ImmutableBytesWritable类型，Value类型为Put。使用connection工厂创建与HBase集群连接，跟据表target_table的名称获取表对象，插入数据并关闭表之后关闭连接。
4. 运行程序
我们可以通过以下命令运行程序：
hadoop jar copy-hbase-data-1.0.jar com.example.CopyHBaseData
其中，copy-hbase-data-1.0.jar是项目构建后的jar包文件名。可以在Hadoop的日志信息中，查看MapReduce程序的执行情况。
程序执行完成后，就会将HBase表中的数据成功地复制到我们的目标表中。
5. 总结
本文介绍如何使用通用的MapReduce程序，在Hadoop集群中对HBase表数据进行复制。在实际应用中，我们可以根据需求修改程序的代码，实现不同的数据输出方式，来满足不同的需求。