mapreduce中實現對hbase中表數據的添加
參考網址：http://www.javabloger.com/article/hadoop-mapreduce-hbase.html
根據參考網址中的小實例，自己親自實現了一下，記錄一下自己對該程序的一些理解。
實例：先將數據文件上傳到HDFS，然后用MapReduce進行處理，將處理后的數據插入到hbase中。代碼如下：
首先是Mapper：
package txt_to_hbase;
import java.io.IOException;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class THMapper extends Mapper<LongWritable,Text,Text,Text>{
public void map(LongWritable key,Text value,Context context){
String[] items = value.toString().split(" ");
String k = items[0];
String v = items[1];
System.out.println("key:"+k+","+"value:"+v);
try {
context.write(new Text(k), new Text(v));
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
然后是Reduce：
package txt_to_hbase;
import java.io.IOException;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.io.Text;
public class THReducer extends TableReducer<Text,Text,ImmutableBytesWritable>{
public void reduce(Text key,Iterable<Text> value,Context context){
String k = key.toString();
String v = value.iterator().next().toString(); //由數據知道value就只有一行
Put putrow = new Put(k.getBytes());
putrow.add("f1".getBytes(), "qualifier".getBytes(), v.getBytes());
try {
context.write(new ImmutableBytesWritable(key.getBytes()), putrow);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
然后是Driver：
package txt_to_hbase;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.util.Tool;
public class THDriver extends Configured implements Tool{
@Override
public int run(String[] arg0) throws Exception {
// TODO Auto-generated method stub
Configuration conf = HBaseConfiguration.create();
conf.set("hbase.zookeeper.quorum.", "localhost"); //千萬別忘記配置
Job job = new Job(conf,"Txt-to-Hbase");
job.setJarByClass(TxtHbase.class);
Path in = new Path("/home/daisy/inout/txthbase/");
job.setInputFormatClass(TextInputFormat.class);
FileInputFormat.addInputPath(job, in);
job.setMapperClass(THMapper.class);
job.setReducerClass(THReducer.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Text.class);
TableMapReduceUtil.initTableReducerJob("tab1", THReducer.class, job);
job.waitForCompletion(true);
return 0;
}
}
最后是主類：
package txt_to_hbase;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.ToolRunner;
public class TxtHbase {
public static void main(String [] args) throws Exception{
int mr;
mr = ToolRunner.run(new Configuration(),new THDriver(),args);
System.exit(mr);
}
}
輸入文件是3個txt文件，每個txt中的文件內容均是如下格式：
1 name1--txt1-www.javabloger.com
2 name2--txt1
3 name3--txt1
4 name4--txt1
5 name5--txt1
通過以上代碼，mapreduce實現之后，在hbase的shell中查看tab1表，如下：
hbase(main):009:0> scan 'tab1'
ROW COLUMN+CELL
1 column=f1:qualifier, timestamp=1320235555118, value=name1--txt1-www.javabloger.com
10 column=f1:qualifier, timestamp=1320235555118, value=name10--txt2
11 column=f1:qualifier, timestamp=1320235555118, value=name11--txt3-www.javabloger.com
12 column=f1:qualifier, timestamp=1320235555118, value=name12--txt3
13 column=f1:qualifier, timestamp=1320235555118, value=name13--txt3
14 column=f1:qualifier, timestamp=1320235555118, value=name14--txt3
15 column=f1:qualifier, timestamp=1320235555118, value=name15--txt3
2 column=f1:qualifier, timestamp=1320235555118, value=name2--txt1
3 column=f1:qualifier, timestamp=1320235555118, value=name3--txt1
4 column=f1:qualifier, timestamp=1320235555118, value=name4--txt1
5 column=f1:qualifier, timestamp=1320235555118, value=name5--txt1
6 column=f1:qualifier, timestamp=1320235555118, value=name6--txt2-www.javabloger.com
7 column=f1:qualifier, timestamp=1320235555118, value=name7--txt2
8 column=f1:qualifier, timestamp=1320235555118, value=name8--txt2
9 column=f1:qualifier, timestamp=1320235555118, value=name9--txt2
15 row(s) in 0.0570 seconds
Map跟普通的mapreduce函數沒有多大區別，正常的TextInputFormat方式輸入，按行讀取。
Reduce中要把處理之后的結果寫入hbase的表中，所以與普通的mapreduce程序有些區別，由以上代碼可以知道，reduce類繼承的是TableReducer，通過查詢API（如下圖1）知道，它也是一種基本的Reducer類，與其他的reduce類一樣，它的輸入k/v對是對應Map的輸出k/v對，它的輸出key可以是任意的類型，但是value必須是一個put或delete實例。
圖1：TableReducer類詳解
Reduce的輸出key是ImmutableWritable類型（org.apache.hadoop.hase.io），API中的解釋，它是一個可以用作key或value類型的字節序列，該類型基於BytesWritable，不能調整大小。Reduce的輸出value是一個put。如上面代碼： context.write(new ImmutableBytesWritable(key.getBytes()), putrow);
Driver中job配置的時候沒有設置 job.setReduceClass(); 而是用 TableMapReduceUtil.initTableReducerJob("tab1", THReducer.class, job); 來執行reduce類。
TableMapReduceUtil類（org.apache.hadoop.hbase.mapreduce）:a utility for TableMapper or TableReducer。因為本例子中的reduce繼承的是TableReducer，所以也就解釋了用TableMapReduceUtil來執行的原因。該類的方法有：addDependencyJars（），initTableMapperJob（），initTableReducerJob（），limitNumReduceTasks（），setNumReduceTasks（）等，詳細包括參數等可以查看API。
同時注意本程序代碼的格式，將Map，Reduce，以及Job的配置分離，比較清晰。之前寫代碼喜歡把map，reduce 以及job配置全都寫在一個類中，可能這是一種不太好的習慣。這里注意Driver類，要繼承 Configured 類和實現 Tool 接口，以及實現Tool中的run方法，在run方法中對job進行配置。 同時main函數中用ToolRunner.run() 方法來調用Driver類。
本人的一點理解，如有錯誤，歡迎指正，也歡迎大家一起交流mapreduce編程的知識，我的email：dongtingting8877@163.com 。