WordCountHbaseReaderMapper類繼承了TableMapper< Text,Text>抽象類，TableMapper類專門用于完成MapReduce中Map過程與Hbase表之間的操作。此時的map(ImmutableBytesWritable key,Result value,Context context)方法，第一個參數(shù)key為Hbase表的rowkey主鍵，第二個參數(shù)value為key主鍵對應(yīng)的記錄集合，此處的map核心實(shí)現(xiàn)是遍歷key主鍵對應(yīng)的記錄集合value，將其組合成一條記錄通過contentx.write(key,value)填充到< key,value>鍵值對中。
詳細(xì)源碼請參考：WordCountHbaseReader\src\com\zonesion\hbase\WordCountHbaseReader.java
public static class WordCountHbaseReaderMapper extends
TableMapper<Text,Text>{

@Override
protected void map(ImmutableBytesWritable key,Result value,Context context)
throws IOException, InterruptedException {
StringBuffer sb = new StringBuffer("");
for(Entry<byte[],byte[]> entry:value.getFamilyMap("content".getBytes()).entrySet()){
String str = new String(entry.getValue());
//將字節(jié)數(shù)組轉(zhuǎn)換為String類型
if(str != null){
sb.append(new String(entry.getKey()));
sb.append(":");
sb.append(str);
}
context.write(new Text(key.get()), new Text(new String(sb)));
}
}
}

3、 Reducer函數(shù)實(shí)現(xiàn)
此處的WordCountHbaseReaderReduce實(shí)現(xiàn)了直接輸出Map輸出的< key,value>鍵值對，沒有對其做任何處理。詳細(xì)源碼請參考：WordCountHbaseReader\src\com\zonesion\hbase\WordCountHbaseReader.java
public static class WordCountHbaseReaderReduce extends Reducer<Text,Text,Text,Text>{
private Text result = new Text();
@Override
protected void reduce(Text key, Iterable<Text> values,Context context)
throws IOException, InterruptedException {
for(Text val:values){
result.set(val);
context.write(key, result);
}
}
}

4、 驅(qū)動函數(shù)實(shí)現(xiàn)
與WordCount的驅(qū)動類不同，在Job配置的時候沒有配置job.setMapperClass()，而是用以下方法執(zhí)行Mapper類： TableMapReduceUtil.initTableMapperJob(tablename,scan,WordCountHbaseReaderMapper.class, Text.class, Text.class, job);
該方法指明了在執(zhí)行job的Map過程時，數(shù)據(jù)輸入源是hbase的tablename表，通過掃描讀入對象scan對表進(jìn)行全表掃描，為Map過程提供數(shù)據(jù)源輸入，通過WordCountHbaseReaderMapper.class執(zhí)行Map過程，Map過程的輸出key/value類型是 Text.class與Text.class，最后一個參數(shù)是作業(yè)對象。特別注意：這里聲明的是一個最簡單的掃描讀入對象scan，進(jìn)行表掃描讀取數(shù)據(jù)，其中scan可以配置參數(shù)，這里為了例子簡單不再詳述，用戶可自行嘗試。
詳細(xì)源碼請參考：WordCountHbaseReader\src\com\zonesion\hbase\WordCountHbaseReader.java
public static void main(String[] args) throws Exception {
String tablename = "wordcount";
Configuration conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum", "Master");
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length != 1) {
System.err.println("Usage: WordCountHbaseReader <out>");
System.exit(2);
}
Job job = new Job(conf, "WordCountHbaseReader");
job.setJarByClass(WordCountHbaseReader.class);
//設(shè)置任務(wù)數(shù)據(jù)的輸出路徑；
FileOutputFormat.setOutputPath(job, new Path(otherArgs[0]));
job.setReducerClass(WordCountHbaseReaderReduce.class);
Scan scan = new Scan();
TableMapReduceUtil.initTableMapperJob(tablename,scan,WordCountHbaseReaderMapper.class, Text.class, Text.class, job);
//調(diào)用job.waitForCompletion(true) 執(zhí)行任務(wù)，執(zhí)行成功后退出；
System.exit(job.waitForCompletion(true) ? 0 : 1);

}

5、部署運(yùn)行
1）啟動Hadoop集群和Hbase服務(wù)
[hadoop@K-Master ~]$ start-dfs.sh #啟動hadoop HDFS文件管理系統(tǒng)
[hadoop@K-Master ~]$ start-mapred.sh #啟動hadoop MapReduce分布式計(jì)算服務(wù)
[hadoop@K-Master ~]$ start-hbase.sh #啟動Hbase
[hadoop@K-Master ~]$ jps #查看進(jìn)程
22003 HMaster
10611 SecondaryNameNode
22226 Jps
21938 HQuorumPeer
10709 JobTracker
22154 HRegionServer
20277 Main
10432 NameNode