基于HBase的入库草案效率对比验证（二）

上节中，我们采用JavaAPI的方式来操作HBase，接连和访问方式都比较简单直接，而本节我们采用MapReduce的方式来操作HBase，那么就要先配置好Eclipse-Hadoop的插件。

一、安装Eclipse-Hadoop插件

由于网上这方面的资料非常全，所以本人推荐一个参考博文，照着配置就OK：

http://www.cnblogs.com/flyoung2008/archive/2011/12/09/2281400.html

二、定义Map

package txt_to_hbase;

import java.io.IOException;
import java.util.Random;

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) {
        long begintime = System.currentTimeMillis();
        System.out.println("map开始时间："+begintime);
        String[] items = value.toString().split("--");
        String k = String.valueOf((new Random()).nextLong());//生成随机rowkey
        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();
        }
        long endtime = System.currentTimeMillis();
       // System.out.println("map耗时："+(endtime - begintime));
    }
}

 Map跟普通的mapreduce函数没有多大区别，正常的TextInputFormat方式输入，按行读取。

 Reduce中要把处理之后的结果写入hbase的表中，所以与普通的mapreduce程序有些区别，由以上代码可以知道，reduce类继承的是TableReducer，通过查询API（如下图1）知道，它也是一种基本的Reducer类，与其他的reduce类一样，它的输入k/v对是对应Map的输出k/v对，它的输出key可以是任意的类型，但是value必须是一个put或delete实例。

三、定义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> values, Context context) {
        long begintime = System.currentTimeMillis();
        String k = key.toString();
        String v = values.iterator().next().toString(); // 由数据知道value就只有一行
        Put putrow = new Put(k.getBytes());
        putrow.addColumn("fam1".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();
        }
        long endtime = System.currentTimeMillis();
        if(Jishu.stac==0L) {
            Jishu.stac = System.currentTimeMillis();
        }
        Jishu.endc = System.currentTimeMillis();
        System.out.println("stac="+Jishu.stac+",endc="+Jishu.endc+",total="+(Jishu.endc-Jishu.stac));
    }

}

Reduce的输出key是ImmutableWritable类型（org.apache.hadoop.hase.io），API中的解释，它是一个可以用作key或value类型的字节序列，该类型基于BytesWritable，不能调整大小。Reduce的输出value是一个put。

四、定义drive：

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
        System.out.println("11111111111111111111");
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "192.168.13.74"); 
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        
        Job job = Job.getInstance(conf,"Txt-to-Hbase");
        job.setJarByClass(TxtHbase.class);
        Path in = new Path("hdfs://192.168.13.74:9000/cjt/input");
        job.setInputFormatClass(TextInputFormat.class);
        FileInputFormat.addInputPath(job, in);

        long begintime = System.currentTimeMillis();
        System.out.println("当前时间1："+System.currentTimeMillis());
        job.setMapperClass(THMapper.class);
        job.setReducerClass(THReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        System.out.println("当前时间2："+System.currentTimeMillis());
        TableMapReduceUtil.initTableReducerJob("tab1", THReducer.class, job);

        long endtime = System.currentTimeMillis();
        System.out.println("当前时间3："+System.currentTimeMillis());
       job.waitForCompletion(true);
       return 0;
    }

}

 跟普通的MapReduce操作基本一样，唯一不同的是，在执行Hbase写入操作时使用的是TableMapReduceUtil.initTableReducerJob，同理执行Hbase读取操作应该使用TableMapReduceUtil.initTableMapperJob。而我们的例子中，读取的是文件，非Hbase表数据，所以用到的是：

	Path in = new Path("hdfs://192.168.13.74:9000/cjt/input");
        job.setInputFormatClass(TextInputFormat.class);
        FileInputFormat.addInputPath(job, in);

五、定义主类：

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);
    }
}

ok，至此我们可以运行该测试类，首先我们把数据文件上传到HDFS上，我这边测试了1千，1万，3万，10万，15万，30万，60万条记录的入库速度，结果如下：