hbase的wordcount

Posted 2020-10-16 求知cvip

package com.neworigin.HBaseMR;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Mutation;
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.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

public class HbaseMRTest {
    static Configuration conf=null;
    static{
//        配置configuration的三种方法
//        ①直接将hbase-site.xml拿来放到src下面
        conf=HBaseConfiguration.create();
//        ②设置服务器和端口        
//        conf.set("hbase.zookeeper.quorum", "s100:2181,s101:2181,s102:2181");
//        ③扥开设置服务器和端口
        conf.set("hbase.zookeeper.quorum", "s100,s101,s102");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
    }
    
////    表信息
//    public static final String tablename="wordtest";//表一
//    public static final String colf="content";//表列族
//    public static final String col="info";//列
//    
//    public static final String tablename2="stat";//表二
//    
//public static void initTB(){
//    
//}    
public static class HBmapper extends TableMapper<Text,IntWritable>/*输出类型*/{
    private static IntWritable one=new IntWritable(1);
    private static Text word =new Text();
//    输入类型，key：row key  value：一行数据的结果集 result
    protected void map(ImmutableBytesWritable key, Result value,
            Mapper<ImmutableBytesWritable, Result, Text, IntWritable>.Context context)
            throws IOException, InterruptedException {
        for(Cell cell:value.rawCells())
        {
            word.set(CellUtil.cloneValue(cell));//读取值
            context.write(word, one);//输出：单词----1
        }
        
    }
}
public static class HBreducer extends TableReducer<Text,IntWritable,ImmutableBytesWritable>{

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values,
            Reducer<Text, IntWritable, ImmutableBytesWritable, Mutation>.Context context)
            throws IOException, InterruptedException {
        
        int sum=0;
        //叠加单词个数
        for(IntWritable value:values)
        {
            sum+=value.get();
        }
        Put put = new Put(Bytes.toBytes(key.toString()));//设置row key为单词
        put.add(Bytes.toBytes("content"), Bytes.toBytes("info"), Bytes.toBytes(String.valueOf(sum)));
        //写到hbase中的需要指定rowkey和put
        context.write(new ImmutableBytesWritable(Bytes.toBytes(key.toString())), put);
    }
    
}
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
    Job job=new Job(conf,"HBaseMR");
    Scan scan =new Scan();
    TableMapReduceUtil.initTableMapperJob("wordtest",scan, HBmapper.class, Text.class, IntWritable.class, job);
    TableMapReduceUtil.initTableReducerJob("stat", HBreducer.class, job);
    
    job.waitForCompletion(true);
    System.out.println("finished");
}
}