074 hbase与mapreduce集成

一：运行给定的案例

1.获取jar包里的方法

　　

2.运行hbase自带的mapreduce程序

　　lib/hbase-server-0.98.6-hadoop2.jar　

　

3.具体运行

　　注意命令：mapredcp。

　　HADOOP_CLASSPATH是当前运行时需要的环境。

　　

4.运行一个小方法

　　$HADOOP_HOME/bin/yarn jar lib/hbase-server-0.98.6-hadoop2.jar rowcounter nstest1:tb1

　　

 二：自定义hbase的数据拷贝

1.需求

　　将nstest1:tb1的数据info：name列拷贝到nstest1:tb2

2.新建tb2表

　　

3.书写mapreduce程序

　　输入：rowkey，result。

package com.beifeng.bigdat;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class HBaseMRTest extends Configured implements Tool{
    /**
     * map
     * @author27      *
     */
    public static class tbMap extends TableMapper<ImmutableBytesWritable, Put>{

        @Override
        protected void map(ImmutableBytesWritable key, Result value,Context context) throws IOException, InterruptedException {
            Put put=new Put(key.get());
            for(Cell cell:value.rawCells()){
                if("info".equals(Bytes.toString(CellUtil.cloneFamily(cell)))){
                    if("name".equals(Bytes.toString(CellUtil.cloneQualifier(cell)))){
                        put.add(cell);
                        context.write(key, put);
                    }
                }
            }
        }
        
    }
    /**
     * reduce
     * @author48      *
     */
    public static class tbReduce extends TableReducer<ImmutableBytesWritable, Put, ImmutableBytesWritable>{

        @Override
        protected void reduce(ImmutableBytesWritable key, Iterable<Put> values,Context context)throws IOException, InterruptedException {
            for(Put put:values){
                context.write(key, put);
            }
        }
        
    }
    
    public int run(String[] args) throws Exception {
        Configuration conf=super.getConf();
        Job job =Job.getInstance(conf, "hbasemr");
        job.setJarByClass(HBaseMRTest.class);
        Scan scan=new Scan();
        TableMapReduceUtil.initTableMapperJob(
                "nstest1:tb1",
                scan,
                tbMap.class,
                ImmutableBytesWritable.class,
                Put.class,
                job);
        TableMapReduceUtil.initTableReducerJob(
                "nstest1:tb2",
                tbReduce.class,
                job);
        boolean issucess=job.waitForCompletion(true);
        return issucess?0:1;
    }
    public static void main(String[] args) throws Exception{
        Configuration conf=HBaseConfiguration.create();
        int status=ToolRunner.run(conf, new HBaseMRTest(), args);
        System.exit(status);
    }
    
}

4.打成jar包

5.运行语句

　　加上需要的export前提。

 　　$HADOOP_HOME/bin/yarn jar /etc/opt/datas/HBaseMR.jar com.beifeng.bigdat.HBaseMRTest

6.效果