MapReduce接口
HBase提供了TableInputFormat、TableOutputFormat、TableMapper和TableReducer类来支持使用MapReduce框架处理HBase上的数据，并提供了TableMapReduceUtil类来初始化一个HBase-MapReduce任务。下面介绍一下这些接口。
TableInputFormat类
TableInputFormat负责将HBase数据按Region进行切片，该类继承自TableInputFormatBase类，TableInputFormatBase类实现了InputFormat类的大部分功能，TableInputFormat只是在其上添加了几个配置接口。TableInputFormat类通过setConf接口进行配置。如果需要自定义HBase的InputFormat类，可以通过重载TableInputFormatBase类的方法进行开发。
TableOutputFormat类
TableOutputFormat类负责将MapReduce任务输出的数据写入HBase表中。TableOutputFormat类同样通过setConf方法进行配置，如通过设置 TableOutputFormat.OUTPUT_TABLE来设置输出的目标表格。
TableMapper类
TableMapper类是一个抽象类，继承自Mapper类，如下所示：
|
/* @param <KEYOUT> The type of the key.
* @param <VALUEOUT> The type of the value.
* @see org.apache.hadoop.mapreduce.Mapper
*/
public abstract class TableMapper<KEYOUT, VALUEOUT>
extends Mapper<ImmutableBytesWritable, Result, KEYOUT, VALUEOUT> {
}
|
TableMapper输入的Key为RowKey的字节码数据，输入的Value为Result类型，表示一行数据。开发者需要重载TableMapper类的map方法来实现自己的Map任务。
TableReducer类
TableReducer类也是一个抽象类，继承自Reducer类，如下所示：
|
/* @param <KEYIN> The type of the input key.
* @param <VALUEIN> The type of the input value.
* @param <KEYOUT> The type of the output key.
* @see org.apache.hadoop.mapreduce.Reducer
*/
public abstract class TableReducer<KEYIN, VALUEIN, KEYOUT>
extends Reducer<KEYIN, VALUEIN, KEYOUT, Writable> {
}
|
可见，TableReducer与普通的Reducer类没有区别，开发者需要重载TableReducer类的reduce方法来实现自己的Map任务。
TableMapReduceUtil类
TableMapReduceUtil是一个辅助类，用来简化一个HBase-MapReduce作业的配置过程。该类提供了多个方法来初始化map任务和reduce任务。常见的方法如下 ：
static void initTableMapperJob(byte[] table, Scan scan,
Class<? extends TableMapper> mapper,
Class<?> outputKeyClass,
Class<?> outputValueClass,
org.apache.hadoop.mapreduce.Job job)
static void initTableReducerJob(String table,
Class<? extends TableReducer> reducer,
org.apache.hadoop.mapreduce.Job job)
void setNumReduceTasks(String table,
org.apache.hadoop.mapreduce.Job job)
MapReduce接口示例
下面给出了一个HBase-MapReduce应用实例，该例将一张表中不同的值进行统计，将结果输出到另一张表中。
|
public class HBaseMapReduceDemo {
public static void main(String[] argv){
if(argv.length < 2){
System.exit(0);
}
String sourceTable = argv[0];
String targetTable = argv[1];
Configuration config = HBaseConfiguration.create();
Job job = new Job(config,"ExampleReadWrite");
job.setJarByClass(HBaseMapReduceDemo.class);
Scan scan = new Scan();
scan.setCaching(500); // 在MR作业中，适当设置该值可提升性能
scan.setCacheBlocks(false); // 在MR作业中，应总为false
TableMapReduceUtil.initTableMapperJob(
sourceTable, // 输入表
scan, // 扫描表配置
MyMapper.class, // mapper类
Text.class, // mapper输出Key
IntWritable.class, // mapper输出Value
job);
TableMapReduceUtil.initTableReducerJob(
targetTable, // 输出表
MyTableReducer.class, // reducer类
job);
job.setNumReduceTasks(0);
boolean b = job.waitForCompletion(true);
if (!b) {
throw new IOException("error with job!");
}
}
public static class MyMapper
extends TableMapper<Text, IntWritable> {
private final IntWritable ONE = new IntWritable(1);
private Text text = new Text();
public void map(ImmutableBytesWritable row, Result value, Context context)
throws IOException, InterruptedException {
String val = new String(value.getValue(Bytes.toBytes("cf"),
Bytes.toBytes("attr1")));
text.set(val);
context.write(text, ONE);
}
}
public static class MyTableReducer
extends TableReducer<Text, IntWritable, ImmutableBytesWritable> {
public void reduce(Text key, Iterable<IntWritable> values, Context context)
throws IOException, InterruptedException {
int i = 0;
for (IntWritable val : values) {
i += val.get();
}
Put put = new Put(Bytes.toBytes(key.toString()));
put.add(Bytes.toBytes("cf"), Bytes.toBytes("count"), Bytes.toBytes );
context.write(null, put);
}
}
}
|