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 。