package com.hbase.test;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;

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 {

public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {

Configuration conf = HBaseConfiguration.create();

//配置conf

conf.set("hbase.zookeeper.quorum", "bigdata01,bigdata02,bigdata03");

conf.set("hbase.zookeeper.property.clientPort", "2181");

Job job = Job.getInstance(conf, "word-count");

//指定执行job的主类

job.setJarByClass(HbaseMrTest.class);

Scan scan = new Scan();

//定义mapper需要扫描的列

scan.addColumn(Bytes.toBytes("content"), Bytes.toBytes("words"));

//配置mapper

TableMapReduceUtil.initTableMapperJob("wordcount", scan,HMapper.class , Text.class, IntWritable.class, job);

//配置recuder

TableMapReduceUtil.initTableReducerJob("result", HReducer.class, job);

//提交job

System.exit(job.waitForCompletion(true)?:1);

}

// Text, IntWritable 为输出类型

class HMapper extends TableMapper<Text, IntWritable>{

Text out = new Text();

IntWritable iw = new IntWritable(1);

@Override

protected void map(ImmutableBytesWritable key, Result value,

Mapper<ImmutableBytesWritable, Result, Text, IntWritable>.Context context)

throws IOException, InterruptedException {

//通过result 直接过得content:words 的值

byte[] bytes = value.getValue(Bytes.toBytes("content"), Bytes.toBytes("words"));

if(bytes!=null) {

String words = Bytes.toString(bytes);

//对获得的一行单词进行分割

String[] ws = words.split(" ");

for(String wd : ws) {

out.set(wd);

//写出值，如: you 1

context.write(out, iw);

}

// Text, IntWritable 为mapper的输出类型

class HReducer extends TableReducer<Text, IntWritable, ImmutableBytesWritable>{

@Override

protected void reduce(Text text, Iterable<IntWritable> iter,

Reducer<Text, IntWritable, ImmutableBytesWritable, Mutation>.Context context)

throws IOException, InterruptedException {

int sum = ;

//对iter遍历

for(IntWritable intw : iter) {

sum+= intw.get();

}

//new 一个put 构造函数内的值为row key

Put put = new Put(Bytes.toBytes(text.toString()));

//put添加columnfamily 和column

put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("wordcnt"), Bytes.toBytes(String.valueOf(sum)));

//将每个单词当做row key 写出，put是相加的总和

context.write(new ImmutableBytesWritable(Bytes.toBytes(text.toString())), put);

}

最后将java文件export为RaunableJar放到linux java -jar hbase.jar com.hbase.test.HbaseMrTest 运行

原始数据：

运行结果：

来自 “ ITPUB博客 ” ，链接：http://blog.itpub.net/31506529/viewspace-2214248/，如需转载，请注明出处，否则将追究法律责任。

ITPub博客

Hbase 编写mapreduce

最新文章