MapReducer--从Hbase读取数据

MapReduce输入数据源不仅可以是文件还可以是Hbase，从Hbase读取数据需要注意以下细节

1、Mapper类需要从TableMapper继承，并且实现函数
void map(ImmutableBytesWritable key, Result columns, Context context)
ImmutableBytesWritable key 实际上是Hbase表记录的
rowkeyResult columns 是hbase表记录的字段集合通过如下方式获取字段值，
其中RR表示列簇，
createDate是要获取的字段
String createDate = Bytes.toString(columns.getValue(Bytes.toBytes("RR"), Bytes.toBytes("createDate")));
2、在驱动程序中需要做如下操作
A)、设置hbase地址和端口       
    Configuration conf = HBaseConfiguration.create();
        //设置HbaseIP地址
        conf.set("hbase.zookeeper.quorum", "192.168.10.100");
       //设置Hbase port
    conf.set("hbase.zookeeper.property.clientPort", String.valueOf(2181));
B)、设置扫描器及caching，这样避免MapReduce输出日志过多导致MR失败
Scan scan = new Scan();
String startRowKey = "ROM.01.";
scan.setStartRow(Bytes.toBytes(startRowKey));scan.setCaching(3000);
C)、设置要读取的hbase表、扫描器，Mapper类及输出key，输出valueTableMapReduceUtil.initTableMapperJob("report", scan, DetailMapper.class, Text.class, Text.class, job);

具体实现如下：

点击(此处)折叠或打开

1. import org.apache.commons.cli.*;
   
2. import org.apache.commons.lang3.StringUtils;
   
3. import org.apache.commons.logging.Log;
   
4. import org.apache.commons.logging.LogFactory;
   
5. import org.apache.hadoop.conf.Configuration;
   
6. import org.apache.hadoop.fs.Path;
   
7. import org.apache.hadoop.hbase.HBaseConfiguration;
   
8. import org.apache.hadoop.hbase.client.Result;
   
9. import org.apache.hadoop.hbase.client.Scan;
   
10. import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
    
11. import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
    
12. import org.apache.hadoop.hbase.mapreduce.TableMapper;
    
13. import org.apache.hadoop.hbase.util.Bytes;
    
14. import org.apache.hadoop.io.NullWritable;
    
15. import org.apache.hadoop.io.Text;
    
16. import org.apache.hadoop.mapreduce.Job;
    
17. import org.apache.hadoop.mapreduce.Reducer;
    
18. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
19. import org.apache.hadoop.util.GenericOptionsParser;
    
20. 
    
21. import java.io.IOException;
    
22. 
    
23. 
    
24. public class DetailJob {
    
25.     private static final Log log = LogFactory.getLog(DetailJob.class);
    
26.     private static final String NAME = "DetailJob";
    
27. 
    
28.     public static class DetailMapper extends TableMapper<Text, Text> {
    
29.         private String key="";
    
30.         @Override
    
31.         protected void setup(Context context){
    
32.             key = context.getConfiguration().get("key");
    
33.             System.out.println("setup key:"+key);
    
34.         }
    
35.         @Override
    
36.         protected void map(ImmutableBytesWritable key, Result columns, Context context) throws IOException,
    
37.                 InterruptedException {
    
38.             String emmcId = Bytes.toString(columns.getValue(Bytes.toBytes("RR"), Bytes.toBytes("emmcid")));
    
39.             String cpuId = Bytes.toString(columns.getValue(Bytes.toBytes("RR"), Bytes.toBytes("cpuid")));
    
40.             String osVersion = Bytes.toString(columns.getValue(Bytes.toBytes("RR"), Bytes.toBytes("rom")));
    
41.             String createDate = Bytes.toString(columns.getValue(Bytes.toBytes("RR"), Bytes.toBytes("createDate")));
    
42.             String ram = Bytes.toString(columns.getValue(Bytes.toBytes("RR"), Bytes.toBytes("ram")));
    
43. 
    
44.             StringBuilder sb = new StringBuilder();
    
45.             sb.append(emmcId).append("\t");
    
46.             sb.append(cpuId ).append("\t");
    
47.             sb.append(osVersion).append("\t");
    
48.             sb.append(ram).append("\t");
    
49.             sb.append(createDate);
    
50.             context.write(new Text(cpuId), new Text(sb.toString()));
    
51.         }
    
52.     }
    
53. 
    
54.     public static class DetailReducer extends Reducer<Text, Text,
    
55.             NullWritable,
    
56.             Text> {
    
57.         @Override
    
58.         protected void reduce(Text key, Iterable<Text> values, Context context) throws
    
59.                 IOException, InterruptedException {
    
60.             for (Text val : values) {
    
61.                 context.write(NullWritable.get(), new Text(val.toString()));
    
62.             }
    
63.         }
    
64.     }
    
65. 
    
66.     private static CommandLine parseArgs(String[] args) {
    
67.         Options options = new Options();
    
68.         Option o = new Option("d", "table", true,
    
69.                 "table to read from (must exist)");
    
70.         o.setArgName("table-name");
    
71.         o.setRequired(true);
    
72.         options.addOption(o);
    
73. 
    
74.         o = new Option("o", "output", true,
    
75.                 "the directory to write to");
    
76.         o.setArgName("path-in-HDFS");
    
77.         o.setRequired(true);
    
78.         options.addOption(o);
    
79. 
    
80.         o = new Option("k", "key", true,
    
81.                 "the directory to write to");
    
82.         o.setArgName("path-in-HDFS");
    
83.         o.setRequired(true);
    
84.         options.addOption(o);
    
85. 
    
86.         CommandLineParser parser = new PosixParser();
    
87.         CommandLine cmd = null;
    
88.         try {
    
89.             cmd = parser.parse(options, args);
    
90.         } catch (Exception e) {
    
91.             System.err.println("ERROR: " + e.getMessage() + "\n");
    
92.             HelpFormatter formatter = new HelpFormatter();
    
93.             formatter.printHelp(NAME + " ", options, true);
    
94.             System.exit(-1);
    
95.         }
    
96.         return cmd;
    
97.     }
    
98. 
    
99.     public static void main(String[] args) throws Exception {
    
100.         Configuration conf = HBaseConfiguration.create();
     
101.         //设置HbaseIP地址
     
102.         conf.set("hbase.zookeeper.quorum", "192.168.10.100");
     
103.        //设置Hbase port
     
104.         conf.set("hbase.zookeeper.property.clientPort", String.valueOf(2181));
     
105. 
     
106.         String[] otherArgs =
     
107.                 new GenericOptionsParser(conf, args).getRemainingArgs();
     
108.         CommandLine cmd = parseArgs(otherArgs);
     
109.         String date = cmd.getOptionValue("d");
     
110.         String output = cmd.getOptionValue("o");
     
111. 
     
112.         String key = cmd.getOptionValue("k");
     
113.         conf.set("key",key);
     
114. 
     
115.         //设置Hbase扫描数据的起始row
     
116.        // String date = cmd.getOptionValue("d");
     
117.         Scan scan = new Scan();
     
118.         if (date != null && date.trim().length() > 0) {
     
119.             //String nextDayStr = getNextDayStr(date);
     
120.             String startRowKey = "ROM.01." + date.replace("-", "");
     
121.             //String endRowKey = "ROM.01." + nextDayStr.replace("-", "");
     
122.             scan.setStartRow(Bytes.toBytes(startRowKey));
     
123.            // scan.setStopRow(Bytes.toBytes(endRowKey));
     
124.         }
     
125.         scan.setCaching(3000);
     
126. 
     
127.         Job job = new Job(conf, NAME);
     
128.         job.setJarByClass(DetailJob.class);
     
129.         //设置要读取的hbase表、扫描器，Mapper类及输出key，输出value
     
130.         TableMapReduceUtil.initTableMapperJob("report", scan, DetailMapper.class, Text.class, Text.class, job);
     
131.         job.setReducerClass(DetailReducer.class);
     
132.         job.setOutputKeyClass(NullWritable.class);
     
133.         job.setOutputValueClass(Text.class);
     
134.         job.setNumReduceTasks(1);
     
135. 
     
136.         FileOutputFormat.setOutputPath(job, new Path(output));
     
137.         job.waitForCompletion(true);
     
138.     }