mapreduce读取hbase
package com.sun.hbase;import java.io.IOException;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.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.util.Bytes;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Job;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;import org.apache.hadoop.util.Tool;import org.apache.hadoop.util.ToolRunner;/** * 从HBase数据库中读数据 * @author asheng * 读取时要设置好族名和列名（row34） * 输出结果在："/home/asheng/Hbase/out/"下 */public class ReadDataFromHBase {/** * 使用TableMapper说明是从hbase中读取数据，那么无需设置TableMapper的输入类型，只需设置其输出类型即可TableMapper<Text,Text> * 但是在map方法中要使用相应的类型map(ImmutableBytesWritable row, Result value, Context context) */public static class THMapper extends TableMapper<Text,Text> {private Text text = new Text();public void map(ImmutableBytesWritable row, Result value, Context context) {String row_ = new String(row.get());String val = new String(value.getValue(Bytes.toBytes("f1"), Bytes.toBytes("qualifier"))); text.set(val); System.out.println("key:" + row_ + "," + "value:" + text); try { context.write( new Text(row_),text); } catch (IOException e) { e.printStackTrace();} catch (InterruptedException e) { e.printStackTrace();}} }public static class THDriver extends Configured implements Tool{ @Override public int run(String[] arg0) throws Exception { Configuration conf = HBaseConfiguration.create(); conf.set("hbase.zookeeper.quorum.", "localhost"); Job job = new Job(conf,"Hbase-to-Txt"); job.setJarByClass(ReadDataFromHBase.class); Path out = new Path("/home/asheng/Hbase/out/"); job.setOutputFormatClass(TextOutputFormat.class); FileOutputFormat.setOutputPath(job, out); job.setMapperClass(THMapper.class); Scan scan = new Scan(); scan.setCaching(500); scan.setCacheBlocks(false); TableMapReduceUtil.initTableMapperJob("tab", scan, THMapper.class, Text.class, Text.class, job); job.waitForCompletion(true); return 0; }}public static void main(String [] args) throws Exception{ int mr; mr = ToolRunner.run(new Configuration(),new THDriver(),args); System.exit(mr); }}
?
