source: sample/hadoop-0.16/tw/org/nchc/demo/DemoHBaseSource.java @ 27

Last change on this file since 27 was 27, checked in by waue, 16 years ago

test!

File size: 3.1 KB
RevLine 
[21]1/**
2 * Program: DemoHBaseSource.java
3 * Editor: Waue Chen
4 * From :  NCHC. Taiwn
5 * Last Update Date: 07/02/2008
6 * Re-code from : Cloud9: A MapReduce Library for Hadoop
7 */
8
9package tw.org.nchc.demo;
10
11import java.io.IOException;
12import java.util.Iterator;
13import java.util.StringTokenizer;
14
15import org.apache.hadoop.fs.Path;
16import org.apache.hadoop.hbase.HStoreKey;
17import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
18import org.apache.hadoop.hbase.mapred.TableInputFormat;
19import org.apache.hadoop.hbase.mapred.TableMap;
20import org.apache.hadoop.io.IntWritable;
21import org.apache.hadoop.io.MapWritable;
22import org.apache.hadoop.io.Text;
23import org.apache.hadoop.mapred.JobClient;
24import org.apache.hadoop.mapred.JobConf;
25import org.apache.hadoop.mapred.MapReduceBase;
26import org.apache.hadoop.mapred.OutputCollector;
27import org.apache.hadoop.mapred.Reducer;
28import org.apache.hadoop.mapred.Reporter;
29
30/**
31 *
32 */
33public class DemoHBaseSource {
34
35  // mapper: emits (token, 1) for every word occurrence
36  private static class MapClass extends TableMap<Text, IntWritable> {
37
38    // reuse objects to save overhead of object creation
39    private final static IntWritable one = new IntWritable(1);
40    private final static Text textcol = new Text("default:text");
41    private Text word = new Text();
42
43    public void map(HStoreKey key, MapWritable cols,
44        OutputCollector<Text, IntWritable> output, Reporter reporter)
45        throws IOException {
46
47      String line = Text.decode(((ImmutableBytesWritable) cols
48          .get(textcol)).get());
49
50      StringTokenizer itr = new StringTokenizer(line);
51      while (itr.hasMoreTokens()) {
52        word.set(itr.nextToken());
53        output.collect(word, one);
54      }
55    }
56  }
57
58  // reducer: sums up all the counts
59  private static class ReduceClass extends MapReduceBase implements
60      Reducer<Text, IntWritable, Text, IntWritable> {
61
62    // reuse objects
63    private final static IntWritable SumValue = new IntWritable();
64
65    public void reduce(Text key, Iterator<IntWritable> values,
66        OutputCollector<Text, IntWritable> output, Reporter reporter)
67        throws IOException {
68      // sum up values
69      int sum = 0;
70      while (values.hasNext()) {
71        sum += values.next().get();
72      }
73      SumValue.set(sum);
74      output.collect(key, SumValue);
75    }
76  }
77
78  private DemoHBaseSource() {
79  }
80
81  /**
82   * Runs the demo.
83   */
84  public static void main(String[] args) throws IOException {
85    String outputPath = "sample-counts2";
86
87    int mapTasks = 1;
88    int reduceTasks = 1;
89
90    JobConf conf = new JobConf(DemoHBaseSource.class);
91
92    TableMap.initJob("test", "default:text", MapClass.class, conf);
93
94    conf.setJobName("wordcount");
95
96    conf.setNumMapTasks(mapTasks);
97    conf.setNumReduceTasks(reduceTasks);
98
99    conf.setInputFormat(TableInputFormat.class);
100
101    conf.setOutputKeyClass(Text.class);
102    conf.setOutputValueClass(IntWritable.class);
[27]103    conf.setOutputPath(new Path(outputPath));
[21]104   
105    conf.setMapperClass(MapClass.class);
106    conf.setCombinerClass(ReduceClass.class);
107    conf.setReducerClass(ReduceClass.class);
108
109    JobClient.runJob(conf);
110  }
111}
Note: See TracBrowser for help on using the repository browser.