MapReduce-讀取HBase
阿新 • • 發佈:2018-09-17
run hdf read reduce 數據 use leo keyvalue AMM
MapReduce讀取HBase數據
代碼如下
package com.hbase.mapreduce; 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.Cell; import org.apache.hadoop.hbase.CellUtil; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.KeyValue; 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.TableInputFormat; import org.apache.hadoop.hbase.mapreduce.TableMapper; 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; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; /** * @author:FengZhen * @create:2018年9月17日 * MapReduce讀取HBase中數據 */ public class AnalyzeData extends Configured implements Tool{ private static String addr="HDP233,HDP232,HDP231"; private static String port="2181"; public enum Counters { ROWS, COLS, VALID, ERROR } static class AnalyzeMapper extends TableMapper<Text, IntWritable>{ private IntWritable ONE = new IntWritable(1); @Override protected void map(ImmutableBytesWritable key, Result value, Mapper<ImmutableBytesWritable, Result, Text, IntWritable>.Context context) throws IOException, InterruptedException { context.getCounter(Counters.ROWS).increment(1); try { for (Cell cell : value.listCells()) { context.getCounter(Counters.COLS).increment(1); String hbaseValue = Bytes.toString(CellUtil.cloneValue(cell)); context.write(new Text(hbaseValue), ONE); context.getCounter(Counters.VALID).increment(1); } } catch (Exception e) { e.printStackTrace(); context.getCounter(Counters.ERROR).increment(1); } } } static class AnalyzeReducer extends Reducer<Text, IntWritable, Text, IntWritable>{ @Override protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException { int count = 0; for (IntWritable intWritable : values) { count = count + intWritable.get(); } context.write(key, new IntWritable(count)); } } public int run(String[] arg0) throws Exception { String table = arg0[0]; String column = arg0[1]; String outPath = arg0[2]; Scan scan = new Scan(); if (null != column) { byte[][] colkey = KeyValue.parseColumn(Bytes.toBytes(column)); if (colkey.length > 1) { scan.addColumn(colkey[0], colkey[1]); }else { scan.addFamily(colkey[0]); } } Configuration configuration = HBaseConfiguration.create(); configuration.set("hbase.zookeeper.quorum",addr); configuration.set("hbase.zookeeper.property.clientPort", port); configuration.set(TableInputFormat.INPUT_TABLE, table); Job job = Job.getInstance(configuration); job.setJobName("AnalyzeData"); job.setJarByClass(AnalyzeData.class); job.setMapperClass(AnalyzeMapper.class); job.setInputFormatClass(TableInputFormat.class); TableInputFormat.addColumns(scan, KeyValue.parseColumn(Bytes.toBytes(column))); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); //使用TableMapReduceUtil會報類找不到錯誤 //Caused by: java.lang.ClassNotFoundException: com.yammer.metrics.core.MetricsRegistry //TableMapReduceUtil.initTableMapperJob(table, scan, AnalyzeMapper.class, Text.class, IntWritable.class, job); job.setReducerClass(AnalyzeReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); job.setNumReduceTasks(1); FileOutputFormat.setOutputPath(job, new Path(outPath)); return job.waitForCompletion(true) ? 0 : 1; } public static void main(String[] args) throws Exception { String[] params = new String[] {"test_table_mr","data:info","hdfs://fz/data/fz/output/mrReadHBase"}; int exitCode = ToolRunner.run(new AnalyzeData(), params); System.exit(exitCode); } }
MapReduce-讀取HBase