1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.mapreduce;
20
21 import com.google.common.annotations.VisibleForTesting;
22 import org.apache.hadoop.fs.Path;
23 import org.apache.hadoop.hbase.HRegionInfo;
24 import org.apache.hadoop.hbase.HTableDescriptor;
25 import org.apache.hadoop.hbase.classification.InterfaceAudience;
26 import org.apache.hadoop.hbase.classification.InterfaceStability;
27 import org.apache.hadoop.hbase.client.Result;
28 import org.apache.hadoop.hbase.client.Scan;
29 import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
30 import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
31 import org.apache.hadoop.hbase.util.RegionSplitter;
32 import org.apache.hadoop.io.Writable;
33 import org.apache.hadoop.mapreduce.InputFormat;
34 import org.apache.hadoop.mapreduce.InputSplit;
35 import org.apache.hadoop.mapreduce.Job;
36 import org.apache.hadoop.mapreduce.JobContext;
37 import org.apache.hadoop.mapreduce.RecordReader;
38 import org.apache.hadoop.mapreduce.TaskAttemptContext;
39
40 import java.io.DataInput;
41 import java.io.DataOutput;
42 import java.io.IOException;
43 import java.lang.reflect.Method;
44 import java.util.ArrayList;
45 import java.util.List;
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 @InterfaceAudience.Public
88 @InterfaceStability.Evolving
89 public class TableSnapshotInputFormat extends InputFormat<ImmutableBytesWritable, Result> {
90
91 public static class TableSnapshotRegionSplit extends InputSplit implements Writable {
92 private TableSnapshotInputFormatImpl.InputSplit delegate;
93
94
95 public TableSnapshotRegionSplit() {
96 this.delegate = new TableSnapshotInputFormatImpl.InputSplit();
97 }
98
99 public TableSnapshotRegionSplit(TableSnapshotInputFormatImpl.InputSplit delegate) {
100 this.delegate = delegate;
101 }
102
103 public TableSnapshotRegionSplit(HTableDescriptor htd, HRegionInfo regionInfo,
104 List<String> locations, Scan scan, Path restoreDir) {
105 this.delegate =
106 new TableSnapshotInputFormatImpl.InputSplit(htd, regionInfo, locations, scan, restoreDir);
107 }
108
109 @Override
110 public long getLength() throws IOException, InterruptedException {
111 return delegate.getLength();
112 }
113
114 @Override
115 public String[] getLocations() throws IOException, InterruptedException {
116 return delegate.getLocations();
117 }
118
119 @Override
120 public void write(DataOutput out) throws IOException {
121 delegate.write(out);
122 }
123
124 @Override
125 public void readFields(DataInput in) throws IOException {
126 delegate.readFields(in);
127 }
128
129 TableSnapshotInputFormatImpl.InputSplit getDelegate() {
130 return this.delegate;
131 }
132 }
133
134 @VisibleForTesting
135 static class TableSnapshotRegionRecordReader extends
136 RecordReader<ImmutableBytesWritable, Result> {
137 private TableSnapshotInputFormatImpl.RecordReader delegate =
138 new TableSnapshotInputFormatImpl.RecordReader();
139 private TaskAttemptContext context;
140 private Method getCounter;
141
142 @Override
143 public void initialize(InputSplit split, TaskAttemptContext context) throws IOException,
144 InterruptedException {
145 this.context = context;
146 getCounter = TableRecordReaderImpl.retrieveGetCounterWithStringsParams(context);
147 delegate.initialize(
148 ((TableSnapshotRegionSplit) split).delegate,
149 context.getConfiguration());
150 }
151
152 @Override
153 public boolean nextKeyValue() throws IOException, InterruptedException {
154 boolean result = delegate.nextKeyValue();
155 if (result) {
156 ScanMetrics scanMetrics = delegate.getScanner().getScanMetrics();
157 if (scanMetrics != null && context != null) {
158 TableRecordReaderImpl.updateCounters(scanMetrics, 0, getCounter, context, 0);
159 }
160 }
161 return result;
162 }
163
164 @Override
165 public ImmutableBytesWritable getCurrentKey() throws IOException, InterruptedException {
166 return delegate.getCurrentKey();
167 }
168
169 @Override
170 public Result getCurrentValue() throws IOException, InterruptedException {
171 return delegate.getCurrentValue();
172 }
173
174 @Override
175 public float getProgress() throws IOException, InterruptedException {
176 return delegate.getProgress();
177 }
178
179 @Override
180 public void close() throws IOException {
181 delegate.close();
182 }
183 }
184
185 @Override
186 public RecordReader<ImmutableBytesWritable, Result> createRecordReader(
187 InputSplit split, TaskAttemptContext context) throws IOException {
188 return new TableSnapshotRegionRecordReader();
189 }
190
191 @Override
192 public List<InputSplit> getSplits(JobContext job) throws IOException, InterruptedException {
193 List<InputSplit> results = new ArrayList<InputSplit>();
194 for (TableSnapshotInputFormatImpl.InputSplit split :
195 TableSnapshotInputFormatImpl.getSplits(job.getConfiguration())) {
196 results.add(new TableSnapshotRegionSplit(split));
197 }
198 return results;
199 }
200
201
202
203
204
205
206
207
208
209
210 public static void setInput(Job job, String snapshotName, Path restoreDir)
211 throws IOException {
212 TableSnapshotInputFormatImpl.setInput(job.getConfiguration(), snapshotName, restoreDir);
213 }
214
215
216
217
218
219
220
221
222
223
224
225
226 public static void setInput(Job job, String snapshotName, Path restoreDir,
227 RegionSplitter.SplitAlgorithm splitAlgo, int numSplitsPerRegion) throws IOException {
228 TableSnapshotInputFormatImpl.setInput(job.getConfiguration(), snapshotName, restoreDir,
229 splitAlgo, numSplitsPerRegion);
230 }
231 }