利用MapReduce对HBase数据进行统计分析

1、HBase作为一种kv数据库，能够很好的面对高吞吐率的在线数据读写服务，尤其是写操作，但是在非rowkey多条件查询、数据分析、统计等场景下，HBase表现的就不是很好了，这些场景下就比较适合来用MapReduce来计算。2、应用场景假设有一张HBase表article,它有一列是数据来源source，现在需要统计不同来源的文章数量(数据行数)，对于这样的简单统计需求，可以利用MapReduce程序来实现。3、Map程序TableAnalyzeMap继承HBase的TableMapper基类，把source列看做文本，行数自然是整数，所以key-value输出类型自然是<Text,IntWritable>。
[java] view plain copypublic class TableAnalyzeMap extends TableMapper<Text, IntWritable> {  
  
    @Override  
    protected void map(ImmutableBytesWritable key, Result value,  
            Mapper<ImmutableBytesWritable, Result, Text, IntWritable>.Context context)  
            throws IOException, InterruptedException {  
        try {  
            for (Cell cell : value.listCells()) {  
                String qualifier = new String(CellUtil.cloneQualifier(cell));  
                String colValue = new String(CellUtil.cloneValue(cell), "UTF-8");  
                System.out.print(qualifier + "=" + colValue + "\t");  
                context.write(new Text(colValue), new IntWritable(1));  
            }  
        } catch (Exception e) {  
            e.printStackTrace();  
        }  
    }  
  
}  
4、Reduce程序TableAnalyzeReduce继承HBase的TableReducer基类，这里需要把最终聚合后的结果写到目标表中，rowkey对应source，行数对应目标表的column:count列。
[java] view plain copypublic class TableAnalyzeReduce extends TableReducer<Text, IntWritable, ImmutableBytesWritable> {  
  
    @Override  
    protected void reduce(Text key, Iterable<IntWritable> values, Context context)  
            throws IOException, InterruptedException {  
        int i = 0;  
        for (IntWritable val : values) {  
            i += val.get();  
        }  
        Put put = new Put(Bytes.toBytes(key.toString()));  
        put.addColumn(Bytes.toBytes("column"), Bytes.toBytes("count"), Bytes.toBytes(i));  
        System.out.println(key.toString() + "\t" + i);  
        context.write(null, put);  
    }  
  
}  
5、Combiner程序分析map程序，可以看到map阶段处理后的数据是没有任何合并的，key为数据来源source，value都是1，这样的话，如果直接进入reduce阶段，要分发的数据量还是比较大的，会造成网络负担，针对这个问题，可以在map阶段后，做一下本地reduce，这样进入reduce的阶段的数据量会大大减少。
[java] view plain copypublic class TableAnalyzeCombin extends Reducer<Text, IntWritable, Text, IntWritable> {  
  
    @Override  
    protected void reduce(Text key, Iterable<IntWritable> values, Context context)  
            throws IOException, InterruptedException {  
        int i = 0;  
        for (IntWritable val : values) {  
            i += val.get();  
        }  
        context.write(key, new IntWritable(i));  
    }  
  
}  
6、主程序程序接收四个参数，依次为：业务表名称、source字段列族、source字段列名、目标表名，当然也可以使用Apache Commons CLI类解析命令行参数。
scan.setCaching(500)设置每次读取行数，根据实际情况进行配置，scan.setCacheBlocks(false)告诉HBase本次扫描的数据不要放入缓存中。
[java] view plain copypublic static void main(String[] args) throws ClassNotFoundException, InterruptedException {  
    String tableName = args[0];  
    String family = args[1];  
    String column = args[2];  
    String targetTbale = args[3];  
    System.out.println("tableName=" + tableName + ", family=" + family + ", column=" + column + ", targetTbale="  
            + targetTbale);  
  
    Configuration conf = HBaseConfiguration.create();  
    Scan scan = new Scan();  
    scan.setCaching(500);  
    scan.setCacheBlocks(false);  
    scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(column));  
    try {  
        Job job = Job.getInstance(conf, "analyze table data for " + tableName);  
        job.setJarByClass(HBaseMR.class);  
        TableMapReduceUtil.initTableMapperJob(Bytes.toBytes(tableName), scan, TableAnalyzeMap.class, Text.class,  
                IntWritable.class, job);  
        TableMapReduceUtil.initTableReducerJob(targetTbale, TableAnalyzeReduce.class, job);  
        job.setMapperClass(TableAnalyzeMap.class);  
        job.setReducerClass(TableAnalyzeReduce.class);  
        job.setCombinerClass(TableAnalyzeCombin.class);  
        job.setNumReduceTasks(1);  
        System.exit(job.waitForCompletion(true) ? 0 : 1);  
    } catch (IOException e) {  
        e.printStackTrace();  
    }  
}  
7、Job执行整个mapreduce程序开发完成需要生成相应jar包，将jar上传到集群中某个主机上通过hadoop jar运行。
# sudo -u hdfs hadoop jar HBaseMR.jar HBaseMR article column source analyze_result
当然，如果要在hbase集群外执行的话，作业执行程序里就要配置hbase必要环境信息。
在运行时要是出现找不到HBase相关类，需要在将/usr/lib/hbase/lib/下的相关jar软连接到/usr/lib/hadoop-mapreduce/目录下
# ln -s /usr/lib/hbase/lib/*.jar /usr/lib/hadoop-mapreduce/*.jar