跟Hadoop的無縫集成使得使用MapReduce對HBase的數(shù)據(jù)進(jìn)行分布式計(jì)算非常方便，本文將以前面的blog示例，介紹HBase下MapReduce開發(fā)要點(diǎn)。很好理解本文前提是你對Hadoop MapReduce有一定的了解。
HBase MapReduce核心類介紹
首先一起來回顧下MapReduce的基本編程模型，
可以看到最基本的是通過Mapper和Reducer來處理KV對，Mapper的輸出經(jīng)Shuffle及Sort后變?yōu)镽educer的輸入。除了Mapper和Reducer外，另外兩個重要的概念是InputFormat和OutputFormat，定義了Map-Reduce的輸入和輸出相關(guān)的東西。HBase通過對這些類的擴(kuò)展（繼承）來方便MapReduce任務(wù)來讀寫HTable中的數(shù)據(jù)。
實(shí)例分析
我們還是以最初的blog例子來進(jìn)行示例分析，業(yè)務(wù)需求是這樣：找到具有相同興趣的人，我們簡單定義為如果author之間article的tag相同，則認(rèn)為兩者有相同興趣，將分析結(jié)果保存到HBase。除了上面介紹的blog表外，我們新增一張表tag_friend，RowKey為tag，Value為authors,大概就下面這樣。
我們省略了一些跟分析無關(guān)的Column數(shù)據(jù)，上面的數(shù)據(jù)按前面描述的業(yè)務(wù)需求經(jīng)過MapReduce分析，應(yīng)該得到下面的結(jié)果
實(shí)際的運(yùn)算過程分析如下
代碼實(shí)現(xiàn)
有了上面的分析，代碼實(shí)現(xiàn)就比較簡單了。只需以下幾步
定義Mapper類繼承TableMapper，map的輸入輸出KV跟上面的分析一致。public static class Mapper extends TableMapper {
public Mapper() {}
@Override
public void map(ImmutableBytesWritable row, Result values,Context context) throws IOException {
ImmutableBytesWritable value = null;
String[] tags = null;
for (KeyValue kv : values.list()) {
if ("author".equals(Bytes.toString(kv.getFamily()))
&& "nickname".equals(Bytes.toString(kv.getQualifier()))) {
value = new ImmutableBytesWritable(kv.getValue());
}
if ("article".equals(Bytes.toString(kv.getFamily()))
&& "tags".equals(Bytes.toString(kv.getQualifier()))) {
tags = Bytes.toString(kv.getValue()).split(",");
}
}
for (int i = 0; i < tags.length; i++) {
ImmutableBytesWritable key = new ImmutableBytesWritable(
Bytes.toBytes(tags[i].toLowerCase()));
try {
context.write(key,value);
} catch (InterruptedException e) {
throw new IOException(e);
}
}
}
}
復(fù)制代碼
定義Reducer類繼承TableReducer，reduce的輸入輸出KV跟上面分析的一致。public static class Reducer extends TableReducer {
@Override
public void reduce(ImmutableBytesWritable key,Iterable values,
Context context) throws IOException, InterruptedException {
String friends="";
for (ImmutableBytesWritable val : values) {
friends += (friends.length()>0?",":"")+Bytes.toString(val.get());
}
Put put = new Put(key.get());
put.add(Bytes.toBytes("person"), Bytes.toBytes("nicknames"),
Bytes.toBytes(friends));
context.write(key, put);
}
}
復(fù)制代碼
在提交作業(yè)時設(shè)置inputFormat為TableInputFormat,設(shè)置outputFormat為TableOutputFormat，可以借助TableMapReduceUtil類來簡化編碼。public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
conf = HBaseConfiguration.create(conf);
Job job = new Job(conf, "HBase_FindFriend");
job.setJarByClass(FindFriend.class);
Scan scan = new Scan();
scan.addColumn(Bytes.toBytes("author"),Bytes.toBytes("nickname"));
scan.addColumn(Bytes.toBytes("article"),Bytes.toBytes("tags"));
TableMapReduceUtil.initTableMapperJob("blog", scan,FindFriend.Mapper.class,
ImmutableBytesWritable.class, ImmutableBytesWritable.class, job);
TableMapReduceUtil.initTableReducerJob("tag_friend",FindFriend.Reducer.class, job);
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
復(fù)制代碼
對大數(shù)據(jù)感興趣的朋友可以加我的群? 615997810? 一起交流學(xué)習(xí)，還有免費(fèi)資料可以領(lǐng)取?1,_推薦系統(tǒng)理論與實(shí)戰(zhàn)項(xiàng)目 Part2
2,推薦系統(tǒng)理論與實(shí)戰(zhàn) 項(xiàng)目Part1
3.實(shí)時交易監(jiān)控系統(tǒng)項(xiàng)目(下)
4,實(shí)時交易監(jiān)控系統(tǒng)項(xiàng)目(上)
5,用戶行為分析系統(tǒng)項(xiàng)目
6,分布式全文搜索引擎ElasticSearch Part2
7,大數(shù)據(jù)批處理之HIVE詳解
8,ES公開課 part1
9,spark_streaming_
10,數(shù)據(jù)倉庫搭建詳解
11,大數(shù)據(jù)任務(wù)調(diào)度
12,流數(shù)據(jù)集成神器Kafka
13,Spark 公開課
14,海量日志收集利器：Flume
15,Impala簡介
16,Hive簡介
17,MapReduce簡介
18海量數(shù)據(jù)高速存取數(shù)據(jù)庫 HBase
19,淺談Hadoop管理器yarn原理
小結(jié)
本文通過實(shí)例分析演示了使用MapReduce分析HBase的數(shù)據(jù)，需要注意的這只是一種常規(guī)的方式（分析表中的數(shù)據(jù)存到另外的表中），實(shí)際上不局限于此，不過其他方式跟此類似。如果你進(jìn)行到這里，你肯定想要馬上運(yùn)行它看看結(jié)果，希望大家多多關(guān)注哦。
