1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.chukwa.datacollection.writer.hbase;
20
21 import java.io.IOException;
22 import java.util.ArrayList;
23 import java.util.List;
24
25 import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecord;
26 import org.apache.hadoop.chukwa.extraction.engine.ChukwaRecordKey;
27 import org.apache.hadoop.hbase.client.Put;
28
29 public class OutputCollector implements
30 org.apache.hadoop.mapred.OutputCollector<ChukwaRecordKey, ChukwaRecord> {
31
32 private List<Put> buffers;
33 private StringBuffer s = new StringBuffer();
34 private byte[] rowKey = null;
35 private byte[] cf = null;
36 private long now = 0L;
37
38 public OutputCollector() {
39 buffers = new ArrayList<Put>();
40 }
41
42 @Override
43 public void collect(ChukwaRecordKey key, ChukwaRecord value) throws IOException {
44 String[] keyParts = key.getKey().split("/");
45 s.setLength(0);
46 s.append(keyParts[2]);
47 s.append("-");
48 s.append(keyParts[1]);
49
50 rowKey = s.toString().getBytes();
51
52 cf = key.getReduceType().getBytes();
53 now = value.getTime();
54
55 Put kv = new Put(rowKey);
56 for(String field : value.getFields()) {
57 kv.add(cf, field.getBytes(), now , value.getValue(field).getBytes());
58 }
59 buffers.add(kv);
60 }
61
62 public List<Put> getKeyValues() {
63 return buffers;
64 }
65
66 public void clear() {
67 s.setLength(0);
68 rowKey = null;
69 cf = null;
70 buffers.clear();
71 }
72
73 }