Mapreduce读取Hbase表，写数据到一个Hbase表中

public class LabelJob

{

    public static void main(String[] args)

        throws Exception

    {

        Job job = Job.getInstance(new Configuration());

        job.setJarByClass(LabelJob.class);

        job.setJobName("Hbase.LabelJob");

        Configuration conf = job.getConfiguration();

        conf.set("tablename", "product_tags");

        Scan scan = new Scan();

        scan.setCaching(500);

        scan.setCacheBlocks(false);

        //输入表

        TableMapReduceUtil.initTableMapperJob("tb_user", scan, LabelMapper.class, Text.class, Text.class, job);

        job.setReducerClass(LabelReducer.class);

        //输出表

        TableMapReduceUtil.initTableReducerJob("usertags", LabelReducer.class, job);

        job.waitForCompletion(true);

    }

}

　　

public class LabelMapper extends TableMapper<Text, Text>

{

      protected void setup(Context context)

        throws IOException, InterruptedException

    {

        super.setup(context);

        String tablename = context.getConfiguration().get("tablename");

         .................

    }

 protected void map(ImmutableBytesWritable rowKey, Result result, Context context)

        throws IOException, InterruptedException

    {

          String userid = Bytes.toString(rowKey.get()); // 读取HBase用户表rowkey

          String strlabel = fhb.getStringValue(result, "labels", "label");

         String[] userLabels = strlabel.split(",");

....................

}

}

　　

public class LabelReducer extends TableReducer<Text, Text, ImmutableBytesWritable>

{

     @Override

    public void reduce(Text key, Iterable<Text> values, Context context)

        throws IOException, InterruptedException

    {

String rowKey = key.toString();// 读取Map输出

 for (Text v : values)

            {

                String tag = v.toString();

                Long count = tagMap.get(tag);

                tagMap.put(tag, (count == null) ? 1 : (count + 1));// 计数

            }

 Put put = new Put(productId.getBytes());

  put.add("prodtags".getBytes(), "prodtags".getBytes(),outputlabel.toString().getBytes());

context.write(new ImmutableBytesWritable(productId.getBytes()), put);

}

}