View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
18   */
19  package org.apache.hadoop.hbase.mapreduce;
20  
21  import java.io.Closeable;
22  import java.io.IOException;
23  import java.net.InetAddress;
24  import java.net.InetSocketAddress;
25  import java.net.UnknownHostException;
26  import java.util.ArrayList;
27  import java.util.HashMap;
28  import java.util.List;
29  
30  import javax.naming.NamingException;
31  
32  import org.apache.commons.logging.Log;
33  import org.apache.commons.logging.LogFactory;
34  import org.apache.hadoop.conf.Configuration;
35  import org.apache.hadoop.hbase.classification.InterfaceAudience;
36  import org.apache.hadoop.hbase.classification.InterfaceStability;
37  import org.apache.hadoop.hbase.HConstants;
38  import org.apache.hadoop.hbase.HRegionLocation;
39  import org.apache.hadoop.hbase.TableName;
40  import org.apache.hadoop.hbase.client.Admin;
41  import org.apache.hadoop.hbase.client.Connection;
42  import org.apache.hadoop.hbase.client.HTable;
43  import org.apache.hadoop.hbase.client.RegionLocator;
44  import org.apache.hadoop.hbase.client.Result;
45  import org.apache.hadoop.hbase.client.Scan;
46  import org.apache.hadoop.hbase.client.Table;
47  import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
48  import org.apache.hadoop.hbase.util.Addressing;
49  import org.apache.hadoop.hbase.util.Bytes;
50  import org.apache.hadoop.hbase.util.Pair;
51  import org.apache.hadoop.hbase.util.RegionSizeCalculator;
52  import org.apache.hadoop.hbase.util.Strings;
53  import org.apache.hadoop.mapreduce.InputFormat;
54  import org.apache.hadoop.mapreduce.InputSplit;
55  import org.apache.hadoop.mapreduce.JobContext;
56  import org.apache.hadoop.mapreduce.RecordReader;
57  import org.apache.hadoop.mapreduce.TaskAttemptContext;
58  import org.apache.hadoop.net.DNS;
59  import org.apache.hadoop.util.StringUtils;
60  
61  /**
62   * A base for {@link TableInputFormat}s. Receives a {@link Connection}, a {@link TableName},
63   * an {@link Scan} instance that defines the input columns etc. Subclasses may use
64   * other TableRecordReader implementations.
65   *
66   * Subclasses MUST ensure initializeTable(Connection, TableName) is called for an instance to
67   * function properly. Each of the entry points to this class used by the MapReduce framework,
68   * {@link #createRecordReader(InputSplit, TaskAttemptContext)} and {@link #getSplits(JobContext)},
69   * will call {@link #initialize(JobContext)} as a convenient centralized location to handle
70   * retrieving the necessary configuration information. If your subclass overrides either of these
71   * methods, either call the parent version or call initialize yourself.
72   *
73   * <p>
74   * An example of a subclass:
75   * <pre>
76   *   class ExampleTIF extends TableInputFormatBase {
77   *
78   *     {@literal @}Override
79   *     protected void initialize(JobContext context) throws IOException {
80   *       // We are responsible for the lifecycle of this connection until we hand it over in
81   *       // initializeTable.
82   *       Connection connection = ConnectionFactory.createConnection(HBaseConfiguration.create(
83   *              job.getConfiguration()));
84   *       TableName tableName = TableName.valueOf("exampleTable");
85   *       // mandatory. once passed here, TableInputFormatBase will handle closing the connection.
86   *       initializeTable(connection, tableName);
87   *       byte[][] inputColumns = new byte [][] { Bytes.toBytes("columnA"),
88   *         Bytes.toBytes("columnB") };
89   *       // optional, by default we'll get everything for the table.
90   *       Scan scan = new Scan();
91   *       for (byte[] family : inputColumns) {
92   *         scan.addFamily(family);
93   *       }
94   *       Filter exampleFilter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator("aa.*"));
95   *       scan.setFilter(exampleFilter);
96   *       setScan(scan);
97   *     }
98   *   }
99   * </pre>
100  */
101 @InterfaceAudience.Public
102 @InterfaceStability.Stable
103 public abstract class TableInputFormatBase
104 extends InputFormat<ImmutableBytesWritable, Result> {
105 
106   /** Specify if we enable auto-balance for input in M/R jobs.*/
107   public static final String MAPREDUCE_INPUT_AUTOBALANCE = "hbase.mapreduce.input.autobalance";
108   /** Specify if ratio for data skew in M/R jobs, it goes well with the enabling hbase.mapreduce
109    * .input.autobalance property.*/
110   public static final String INPUT_AUTOBALANCE_MAXSKEWRATIO = "hbase.mapreduce.input.autobalance" +
111           ".maxskewratio";
112   /** Specify if the row key in table is text (ASCII between 32~126),
113    * default is true. False means the table is using binary row key*/
114   public static final String TABLE_ROW_TEXTKEY = "hbase.table.row.textkey";
115 
116   private static final Log LOG = LogFactory.getLog(TableInputFormatBase.class);
117 
118   private static final String NOT_INITIALIZED = "The input format instance has not been properly " +
119       "initialized. Ensure you call initializeTable either in your constructor or initialize " +
120       "method";
121   private static final String INITIALIZATION_ERROR = "Cannot create a record reader because of a" +
122             " previous error. Please look at the previous logs lines from" +
123             " the task's full log for more details.";
124 
125   /** Holds the details for the internal scanner.
126    *
127    * @see Scan */
128   private Scan scan = null;
129   /** The {@link Admin}. */
130   private Admin admin;
131   /** The {@link Table} to scan. */
132   private Table table;
133   /** The {@link RegionLocator} of the table. */
134   private RegionLocator regionLocator;
135   /** The reader scanning the table, can be a custom one. */
136   private TableRecordReader tableRecordReader = null;
137   /** The underlying {@link Connection} of the table. */
138   private Connection connection;
139 
140   
141   /** The reverse DNS lookup cache mapping: IPAddress => HostName */
142   private HashMap<InetAddress, String> reverseDNSCacheMap =
143     new HashMap<InetAddress, String>();
144 
145   /**
146    * Builds a {@link TableRecordReader}. If no {@link TableRecordReader} was provided, uses
147    * the default.
148    *
149    * @param split  The split to work with.
150    * @param context  The current context.
151    * @return The newly created record reader.
152    * @throws IOException When creating the reader fails.
153    * @see org.apache.hadoop.mapreduce.InputFormat#createRecordReader(
154    *   org.apache.hadoop.mapreduce.InputSplit,
155    *   org.apache.hadoop.mapreduce.TaskAttemptContext)
156    */
157   @Override
158   public RecordReader<ImmutableBytesWritable, Result> createRecordReader(
159       InputSplit split, TaskAttemptContext context)
160   throws IOException {
161     // Just in case a subclass is relying on JobConfigurable magic.
162     if (table == null) {
163       initialize(context);
164     }
165     // null check in case our child overrides getTable to not throw.
166     try {
167       if (getTable() == null) {
168         // initialize() must not have been implemented in the subclass.
169         throw new IOException(INITIALIZATION_ERROR);
170       }
171     } catch (IllegalStateException exception) {
172       throw new IOException(INITIALIZATION_ERROR, exception);
173     }
174     TableSplit tSplit = (TableSplit) split;
175     LOG.info("Input split length: " + StringUtils.humanReadableInt(tSplit.getLength()) + " bytes.");
176     final TableRecordReader trr =
177         this.tableRecordReader != null ? this.tableRecordReader : new TableRecordReader();
178     Scan sc = new Scan(this.scan);
179     sc.setStartRow(tSplit.getStartRow());
180     sc.setStopRow(tSplit.getEndRow());
181     trr.setScan(sc);
182     trr.setTable(getTable());
183     return new RecordReader<ImmutableBytesWritable, Result>() {
184 
185       @Override
186       public void close() throws IOException {
187         trr.close();
188         closeTable();
189       }
190 
191       @Override
192       public ImmutableBytesWritable getCurrentKey() throws IOException, InterruptedException {
193         return trr.getCurrentKey();
194       }
195 
196       @Override
197       public Result getCurrentValue() throws IOException, InterruptedException {
198         return trr.getCurrentValue();
199       }
200 
201       @Override
202       public float getProgress() throws IOException, InterruptedException {
203         return trr.getProgress();
204       }
205 
206       @Override
207       public void initialize(InputSplit inputsplit, TaskAttemptContext context) throws IOException,
208           InterruptedException {
209         trr.initialize(inputsplit, context);
210       }
211 
212       @Override
213       public boolean nextKeyValue() throws IOException, InterruptedException {
214         return trr.nextKeyValue();
215       }
216     };
217   }
218 
219   protected Pair<byte[][],byte[][]> getStartEndKeys() throws IOException {
220     return getRegionLocator().getStartEndKeys();
221   }
222 
223   /**
224    * Calculates the splits that will serve as input for the map tasks. The
225    * number of splits matches the number of regions in a table.
226    *
227    * @param context  The current job context.
228    * @return The list of input splits.
229    * @throws IOException When creating the list of splits fails.
230    * @see org.apache.hadoop.mapreduce.InputFormat#getSplits(
231    *   org.apache.hadoop.mapreduce.JobContext)
232    */
233   @Override
234   public List<InputSplit> getSplits(JobContext context) throws IOException {
235     boolean closeOnFinish = false;
236 
237     // Just in case a subclass is relying on JobConfigurable magic.
238     if (table == null) {
239       initialize(context);
240       closeOnFinish = true;
241     }
242 
243     // null check in case our child overrides getTable to not throw.
244     try {
245       if (getTable() == null) {
246         // initialize() must not have been implemented in the subclass.
247         throw new IOException(INITIALIZATION_ERROR);
248       }
249     } catch (IllegalStateException exception) {
250       throw new IOException(INITIALIZATION_ERROR, exception);
251     }
252 
253     try {
254       RegionSizeCalculator sizeCalculator =
255           new RegionSizeCalculator(getRegionLocator(), getAdmin());
256       
257       TableName tableName = getTable().getName();
258   
259       Pair<byte[][], byte[][]> keys = getStartEndKeys();
260       if (keys == null || keys.getFirst() == null ||
261           keys.getFirst().length == 0) {
262         HRegionLocation regLoc =
263             getRegionLocator().getRegionLocation(HConstants.EMPTY_BYTE_ARRAY, false);
264         if (null == regLoc) {
265           throw new IOException("Expecting at least one region.");
266         }
267         List<InputSplit> splits = new ArrayList<InputSplit>(1);
268         long regionSize = sizeCalculator.getRegionSize(regLoc.getRegionInfo().getRegionName());
269         TableSplit split = new TableSplit(tableName,
270             HConstants.EMPTY_BYTE_ARRAY, HConstants.EMPTY_BYTE_ARRAY, regLoc
271                 .getHostnamePort().split(Addressing.HOSTNAME_PORT_SEPARATOR)[0], regionSize);
272         splits.add(split);
273         return splits;
274       }
275       List<InputSplit> splits = new ArrayList<InputSplit>(keys.getFirst().length);
276       for (int i = 0; i < keys.getFirst().length; i++) {
277         if (!includeRegionInSplit(keys.getFirst()[i], keys.getSecond()[i])) {
278           continue;
279         }
280         HRegionLocation location = getRegionLocator().getRegionLocation(keys.getFirst()[i], false);
281         // The below InetSocketAddress creation does a name resolution.
282         InetSocketAddress isa = new InetSocketAddress(location.getHostname(), location.getPort());
283         if (isa.isUnresolved()) {
284           LOG.warn("Failed resolve " + isa);
285         }
286         InetAddress regionAddress = isa.getAddress();
287         String regionLocation;
288         try {
289           regionLocation = reverseDNS(regionAddress);
290         } catch (NamingException e) {
291           LOG.warn("Cannot resolve the host name for " + regionAddress + " because of " + e);
292           regionLocation = location.getHostname();
293         }
294   
295         byte[] startRow = scan.getStartRow();
296         byte[] stopRow = scan.getStopRow();
297         // determine if the given start an stop key fall into the region
298         if ((startRow.length == 0 || keys.getSecond()[i].length == 0 ||
299             Bytes.compareTo(startRow, keys.getSecond()[i]) < 0) &&
300             (stopRow.length == 0 ||
301              Bytes.compareTo(stopRow, keys.getFirst()[i]) > 0)) {
302           byte[] splitStart = startRow.length == 0 ||
303             Bytes.compareTo(keys.getFirst()[i], startRow) >= 0 ?
304               keys.getFirst()[i] : startRow;
305           byte[] splitStop = (stopRow.length == 0 ||
306             Bytes.compareTo(keys.getSecond()[i], stopRow) <= 0) &&
307             keys.getSecond()[i].length > 0 ?
308               keys.getSecond()[i] : stopRow;
309   
310           byte[] regionName = location.getRegionInfo().getRegionName();
311           long regionSize = sizeCalculator.getRegionSize(regionName);
312           TableSplit split = new TableSplit(tableName,
313             splitStart, splitStop, regionLocation, regionSize);
314           splits.add(split);
315           if (LOG.isDebugEnabled()) {
316             LOG.debug("getSplits: split -> " + i + " -> " + split);
317           }
318         }
319       }
320       //The default value of "hbase.mapreduce.input.autobalance" is false, which means not enabled.
321       boolean enableAutoBalance = context.getConfiguration()
322         .getBoolean(MAPREDUCE_INPUT_AUTOBALANCE, false);
323       if (enableAutoBalance) {
324         long totalRegionSize=0;
325         for (int i = 0; i < splits.size(); i++){
326           TableSplit ts = (TableSplit)splits.get(i);
327           totalRegionSize += ts.getLength();
328         }
329         long averageRegionSize = totalRegionSize / splits.size();
330         // the averageRegionSize must be positive.
331         if (averageRegionSize <= 0) {
332             LOG.warn("The averageRegionSize is not positive: "+ averageRegionSize + ", " +
333                     "set it to 1.");
334             averageRegionSize = 1;
335         }
336         return calculateRebalancedSplits(splits, context, averageRegionSize);
337       } else {
338         return splits;
339       }
340     } finally {
341       if (closeOnFinish) {
342         closeTable();
343       }
344     }
345   }
346 
347   /**
348    * @deprecated mistakenly made public in 0.98.7. scope will change to package-private
349    */
350   @Deprecated
351   public String reverseDNS(InetAddress ipAddress) throws NamingException, UnknownHostException {
352     String hostName = this.reverseDNSCacheMap.get(ipAddress);
353     if (hostName == null) {
354       String ipAddressString = null;
355       try {
356         ipAddressString = DNS.reverseDns(ipAddress, null);
357       } catch (Exception e) {
358         // We can use InetAddress in case the jndi failed to pull up the reverse DNS entry from the
359         // name service. Also, in case of ipv6, we need to use the InetAddress since resolving
360         // reverse DNS using jndi doesn't work well with ipv6 addresses.
361         ipAddressString = InetAddress.getByName(ipAddress.getHostAddress()).getHostName();
362       }
363       if (ipAddressString == null) throw new UnknownHostException("No host found for " + ipAddress);
364       hostName = Strings.domainNamePointerToHostName(ipAddressString);
365       this.reverseDNSCacheMap.put(ipAddress, hostName);
366     }
367     return hostName;
368   }
369 
370   /**
371    * Calculates the number of MapReduce input splits for the map tasks. The number of
372    * MapReduce input splits depends on the average region size and the "data skew ratio" user set in
373    * configuration.
374    *
375    * @param list  The list of input splits before balance.
376    * @param context  The current job context.
377    * @param average  The average size of all regions .
378    * @return The list of input splits.
379    * @throws IOException When creating the list of splits fails.
380    * @see org.apache.hadoop.mapreduce.InputFormat#getSplits(
381    *   org.apache.hadoop.mapreduce.JobContext)
382    */
383   private List<InputSplit> calculateRebalancedSplits(List<InputSplit> list, JobContext context,
384                                                long average) throws IOException {
385     List<InputSplit> resultList = new ArrayList<InputSplit>();
386     Configuration conf = context.getConfiguration();
387     //The default data skew ratio is 3
388     long dataSkewRatio = conf.getLong(INPUT_AUTOBALANCE_MAXSKEWRATIO, 3);
389     //It determines which mode to use: text key mode or binary key mode. The default is text mode.
390     boolean isTextKey = context.getConfiguration().getBoolean(TABLE_ROW_TEXTKEY, true);
391     long dataSkewThreshold = dataSkewRatio * average;
392     int count = 0;
393     while (count < list.size()) {
394       TableSplit ts = (TableSplit)list.get(count);
395       TableName tableName = ts.getTable();
396       String regionLocation = ts.getRegionLocation();
397       long regionSize = ts.getLength();
398       if (regionSize >= dataSkewThreshold) {
399         // if the current region size is large than the data skew threshold,
400         // split the region into two MapReduce input splits.
401         byte[] splitKey = getSplitKey(ts.getStartRow(), ts.getEndRow(), isTextKey);
402          //Set the size of child TableSplit as 1/2 of the region size. The exact size of the
403          // MapReduce input splits is not far off.
404         TableSplit t1 = new TableSplit(tableName, ts.getStartRow(), splitKey, regionLocation,
405                 regionSize / 2);
406         TableSplit t2 = new TableSplit(tableName, splitKey, ts.getEndRow(), regionLocation,
407                 regionSize - regionSize / 2);
408         resultList.add(t1);
409         resultList.add(t2);
410         count++;
411       } else if (regionSize >= average) {
412         // if the region size between average size and data skew threshold size,
413         // make this region as one MapReduce input split.
414         resultList.add(ts);
415         count++;
416       } else {
417         // if the total size of several small continuous regions less than the average region size,
418         // combine them into one MapReduce input split.
419         long totalSize = regionSize;
420         byte[] splitStartKey = ts.getStartRow();
421         byte[] splitEndKey = ts.getEndRow();
422         count++;
423         for (; count < list.size(); count++) {
424           TableSplit nextRegion = (TableSplit)list.get(count);
425           long nextRegionSize = nextRegion.getLength();
426           if (totalSize + nextRegionSize <= dataSkewThreshold) {
427             totalSize = totalSize + nextRegionSize;
428             splitEndKey = nextRegion.getEndRow();
429           } else {
430             break;
431           }
432         }
433         TableSplit t = new TableSplit(tableName, splitStartKey, splitEndKey,
434                 regionLocation, totalSize);
435         resultList.add(t);
436       }
437     }
438     return resultList;
439   }
440 
441   /**
442    * select a split point in the region. The selection of the split point is based on an uniform
443    * distribution assumption for the keys in a region.
444    * Here are some examples:
445    * startKey: aaabcdefg  endKey: aaafff    split point: aaad
446    * startKey: 111000  endKey: 1125790    split point: 111b
447    * startKey: 1110  endKey: 1120    split point: 111_
448    * startKey: binary key { 13, -19, 126, 127 }, endKey: binary key { 13, -19, 127, 0 },
449    * split point: binary key { 13, -19, 127, -64 }
450    * Set this function as "public static", make it easier for test.
451    *
452    * @param start Start key of the region
453    * @param end End key of the region
454    * @param isText It determines to use text key mode or binary key mode
455    * @return The split point in the region.
456    */
457   @InterfaceAudience.Private
458   public static byte[] getSplitKey(byte[] start, byte[] end, boolean isText) {
459     byte upperLimitByte;
460     byte lowerLimitByte;
461     //Use text mode or binary mode.
462     if (isText) {
463       //The range of text char set in ASCII is [32,126], the lower limit is space and the upper
464       // limit is '~'.
465       upperLimitByte = '~';
466       lowerLimitByte = ' ';
467     } else {
468       upperLimitByte = Byte.MAX_VALUE;
469       lowerLimitByte = Byte.MIN_VALUE;
470     }
471     // For special case
472     // Example 1 : startkey=null, endkey="hhhqqqwww", splitKey="h"
473     // Example 2 (text key mode): startKey="ffffaaa", endKey=null, splitkey="f~~~~~~"
474     if (start.length == 0 && end.length == 0){
475       return new byte[]{(byte) ((lowerLimitByte + upperLimitByte) / 2)};
476     }
477     if (start.length == 0 && end.length != 0){
478       return new byte[]{ end[0] };
479     }
480     if (start.length != 0 && end.length == 0){
481       byte[] result =new byte[start.length];
482       result[0]=start[0];
483       for (int k = 1; k < start.length; k++){
484           result[k] = upperLimitByte;
485       }
486       return result;
487     }
488     // A list to store bytes in split key
489     List resultBytesList = new ArrayList();
490     int maxLength = start.length > end.length ? start.length : end.length;
491     for (int i = 0; i < maxLength; i++) {
492       //calculate the midpoint byte between the first difference
493       //for example: "11ae" and "11chw", the midpoint is "11b"
494       //another example: "11ae" and "11bhw", the first different byte is 'a' and 'b',
495       // there is no midpoint between 'a' and 'b', so we need to check the next byte.
496       if (start[i] == end[i]) {
497         resultBytesList.add(start[i]);
498         //For special case like: startKey="aaa", endKey="aaaz", splitKey="aaaM"
499         if (i + 1 == start.length) {
500           resultBytesList.add((byte) ((lowerLimitByte + end[i + 1]) / 2));
501           break;
502         }
503       } else {
504         //if the two bytes differ by 1, like ['a','b'], We need to check the next byte to find
505         // the midpoint.
506         if ((int)end[i] - (int)start[i] == 1) {
507           //get next byte after the first difference
508           byte startNextByte = (i + 1 < start.length) ? start[i + 1] : lowerLimitByte;
509           byte endNextByte = (i + 1 < end.length) ? end[i + 1] : lowerLimitByte;
510           int byteRange = (upperLimitByte - startNextByte) + (endNextByte - lowerLimitByte) + 1;
511           int halfRange = byteRange / 2;
512           if ((int)startNextByte + halfRange > (int)upperLimitByte) {
513             resultBytesList.add(end[i]);
514             resultBytesList.add((byte) (startNextByte + halfRange - upperLimitByte +
515                     lowerLimitByte));
516           } else {
517             resultBytesList.add(start[i]);
518             resultBytesList.add((byte) (startNextByte + halfRange));
519           }
520         } else {
521           //calculate the midpoint key by the fist different byte (normal case),
522           // like "11ae" and "11chw", the midpoint is "11b"
523           resultBytesList.add((byte) ((start[i] + end[i]) / 2));
524         }
525         break;
526       }
527     }
528     //transform the List of bytes to byte[]
529     byte[] result = new byte[resultBytesList.size()];
530     for (int k = 0; k < resultBytesList.size(); k++) {
531       result[k] = (byte) resultBytesList.get(k);
532     }
533     return result;
534   }
535 
536   /**
537    * Test if the given region is to be included in the InputSplit while splitting
538    * the regions of a table.
539    * <p>
540    * This optimization is effective when there is a specific reasoning to exclude an entire region from the M-R job,
541    * (and hence, not contributing to the InputSplit), given the start and end keys of the same. <br>
542    * Useful when we need to remember the last-processed top record and revisit the [last, current) interval for M-R processing,
543    * continuously. In addition to reducing InputSplits, reduces the load on the region server as well, due to the ordering of the keys.
544    * <br>
545    * <br>
546    * Note: It is possible that <code>endKey.length() == 0 </code> , for the last (recent) region.
547    * <br>
548    * Override this method, if you want to bulk exclude regions altogether from M-R. By default, no region is excluded( i.e. all regions are included).
549    *
550    *
551    * @param startKey Start key of the region
552    * @param endKey End key of the region
553    * @return true, if this region needs to be included as part of the input (default).
554    *
555    */
556   protected boolean includeRegionInSplit(final byte[] startKey, final byte [] endKey) {
557     return true;
558   }
559 
560   /**
561    * Allows subclasses to get the {@link Table}.
562    *
563    * @deprecated use {@link #getTable()}
564    */
565   @Deprecated
566   protected HTable getHTable() {
567     return (HTable) this.getTable();
568   }
569 
570   /**
571    * Allows subclasses to get the {@link RegionLocator}.
572    */
573   protected RegionLocator getRegionLocator() {
574     if (regionLocator == null) {
575       throw new IllegalStateException(NOT_INITIALIZED);
576     }
577     return regionLocator;
578   }
579   
580   /**
581    * Allows subclasses to get the {@link Table}.
582    */
583   protected Table getTable() {
584     if (table == null) {
585       throw new IllegalStateException(NOT_INITIALIZED);
586     }
587     return table;
588   }
589 
590   /**
591    * Allows subclasses to get the {@link Admin}.
592    */
593   protected Admin getAdmin() {
594     if (admin == null) {
595       throw new IllegalStateException(NOT_INITIALIZED);
596     }
597     return admin;
598   }
599 
600   /**
601    * Allows subclasses to set the {@link HTable}.
602    *
603    * Will attempt to reuse the underlying Connection for our own needs, including
604    * retreiving an Admin interface to the HBase cluster.
605    *
606    * @param table  The table to get the data from.
607    * @throws IOException 
608    * @deprecated Use {@link #initializeTable(Connection, TableName)} instead.
609    */
610   @Deprecated
611   protected void setHTable(HTable table) throws IOException {
612     this.table = table;
613     this.connection = table.getConnection();
614     this.regionLocator = table.getRegionLocator();
615     this.admin = this.connection.getAdmin();
616   }
617 
618   /**
619    * Allows subclasses to initialize the table information.
620    *
621    * @param connection  The Connection to the HBase cluster. MUST be unmanaged. We will close.
622    * @param tableName  The {@link TableName} of the table to process. 
623    * @throws IOException 
624    */
625   protected void initializeTable(Connection connection, TableName tableName) throws IOException {
626     if (this.table != null || this.connection != null) {
627       LOG.warn("initializeTable called multiple times. Overwriting connection and table " +
628           "reference; TableInputFormatBase will not close these old references when done.");
629     }
630     this.table = connection.getTable(tableName);
631     this.regionLocator = connection.getRegionLocator(tableName);
632     this.admin = connection.getAdmin();
633     this.connection = connection;
634   }
635 
636   /**
637    * Gets the scan defining the actual details like columns etc.
638    *
639    * @return The internal scan instance.
640    */
641   public Scan getScan() {
642     if (this.scan == null) this.scan = new Scan();
643     return scan;
644   }
645 
646   /**
647    * Sets the scan defining the actual details like columns etc.
648    *
649    * @param scan  The scan to set.
650    */
651   public void setScan(Scan scan) {
652     this.scan = scan;
653   }
654 
655   /**
656    * Allows subclasses to set the {@link TableRecordReader}.
657    *
658    * @param tableRecordReader A different {@link TableRecordReader}
659    *   implementation.
660    */
661   protected void setTableRecordReader(TableRecordReader tableRecordReader) {
662     this.tableRecordReader = tableRecordReader;
663   }
664   
665   /**
666    * Handle subclass specific set up.
667    * Each of the entry points used by the MapReduce framework,
668    * {@link #createRecordReader(InputSplit, TaskAttemptContext)} and {@link #getSplits(JobContext)},
669    * will call {@link #initialize(JobContext)} as a convenient centralized location to handle
670    * retrieving the necessary configuration information and calling
671    * {@link #initializeTable(Connection, TableName)}.
672    *
673    * Subclasses should implement their initialize call such that it is safe to call multiple times.
674    * The current TableInputFormatBase implementation relies on a non-null table reference to decide
675    * if an initialize call is needed, but this behavior may change in the future. In particular,
676    * it is critical that initializeTable not be called multiple times since this will leak
677    * Connection instances.
678    *
679    */
680   protected void initialize(JobContext context) throws IOException {
681   }
682 
683   /**
684    * Close the Table and related objects that were initialized via
685    * {@link #initializeTable(Connection, TableName)}.
686    *
687    * @throws IOException
688    */
689   protected void closeTable() throws IOException {
690     close(admin, table, regionLocator, connection);
691     admin = null;
692     table = null;
693     regionLocator = null;
694     connection = null;
695   }
696 
697   private void close(Closeable... closables) throws IOException {
698     for (Closeable c : closables) {
699       if(c != null) { c.close(); }
700     }
701   }
702 
703 }