1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.chukwa.extraction.demux.processor;
19
20
21 import java.io.IOException;
22 import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecord;
23 import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecordKey;
24 import org.apache.hadoop.mapred.OutputCollector;
25 import org.apache.hadoop.mapred.Reporter;
26
27 public class ChukwaOutputCollector implements
28 OutputCollector<ChukwaRecordKey, ChukwaRecord> {
29 private OutputCollector<ChukwaRecordKey, ChukwaRecord> outputCollector = null;
30 private Reporter reporter = null;
31 private String groupName = null;
32
33 public ChukwaOutputCollector(
34 String groupName,
35 OutputCollector<ChukwaRecordKey, ChukwaRecord> outputCollector,
36 Reporter reporter) {
37 this.reporter = reporter;
38 this.outputCollector = outputCollector;
39 this.groupName = groupName;
40 }
41
42 @Override
43 public void collect(ChukwaRecordKey key, ChukwaRecord value)
44 throws IOException {
45 this.outputCollector.collect(key, value);
46 reporter.incrCounter(groupName, "total records", 1);
47 reporter.incrCounter(groupName, key.getReduceType() + " records", 1);
48 }
49
50 }