blob: cffedb251f1cef808893f489b60b37865ec18e70 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
|
package parallelai.spyglass.hbase;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Vector;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
public class HBaseRecordReaderRegional extends HBaseRecordReaderBase {
static final Log LOG = LogFactory.getLog(HBaseRecordReaderRegional.class);
private byte[] nextKey = null;
private Vector<List<KeyValue>> resultVector = null;
Map<Long, List<KeyValue>> keyValueMap = null;
private HBaseTableSplitRegional multiSplit = null;
private HBaseTableSplitGranular currentSplit = null;
private HBaseRecordReaderGranular currentRecordReader = null;
public void init(HBaseTableSplitRegional mSplit) throws IOException {
multiSplit = mSplit;
LOG.debug("Creating Multi Split for region location : "
+ multiSplit.getRegionLocation() + " -> " + multiSplit);
setNextSplit();
}
public boolean setNextSplit() throws IOException {
currentSplit = multiSplit.getNextSplit();
LOG.debug("IN: setNextSplit : " + currentSplit );
if( currentSplit != null ) {
setSplitValue(currentSplit);
return true;
} else {
return false;
}
}
private void setRecordReaderParms(HBaseRecordReaderGranular trr, HBaseTableSplitGranular tSplit) throws IOException {
HBaseConfigUtils.setRecordReaderParms(trr, tSplit);
trr.setHTable(htable);
trr.setInputColumns(trrInputColumns);
trr.setRowFilter(trrRowFilter);
trr.init();
}
private void setSplitValue(HBaseTableSplitGranular tSplit) throws IOException {
LOG.debug("IN: setSplitValue : " + tSplit );
if( currentRecordReader != null ) currentRecordReader.close();
currentRecordReader = new HBaseRecordReaderGranular();
setRecordReaderParms(currentRecordReader, currentSplit);
}
@Override
public boolean next(ImmutableBytesWritable ibw, Result result) throws IOException {
boolean nextFlag = currentRecordReader.next(ibw, result);
while(nextFlag == false && multiSplit.hasMoreSplits() ) {
totalPos += currentRecordReader.getPos();
setNextSplit();
nextFlag = currentRecordReader.next(ibw, result);
}
return nextFlag;
}
long totalPos = 0;
@Override
public ImmutableBytesWritable createKey() {
return currentRecordReader.createKey();
}
@Override
public Result createValue() {
return currentRecordReader.createValue();
}
@Override
public long getPos() throws IOException {
long pos = totalPos + currentRecordReader.getPos();
return pos;
}
@Override
public void close() throws IOException {
currentRecordReader.close();
}
@Override
public float getProgress() throws IOException {
// ( current count + percent of next count ) / max count
float prog = ((multiSplit.getCurrSplitCount() + currentRecordReader.getProgress()) / multiSplit.getLength());
return prog;
}
}
|
