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