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.hbase.classification.InterfaceAudience;
35  import org.apache.hadoop.hbase.classification.InterfaceStability;
36  import org.apache.hadoop.hbase.HConstants;
37  import org.apache.hadoop.hbase.HRegionLocation;
38  import org.apache.hadoop.hbase.TableName;
39  import org.apache.hadoop.hbase.client.Admin;
40  import org.apache.hadoop.hbase.client.Connection;
41  import org.apache.hadoop.hbase.client.ConnectionFactory;
42  import org.apache.hadoop.hbase.client.HTable;
43  import org.apache.hadoop.hbase.client.NeedUnmanagedConnectionException;
44  import org.apache.hadoop.hbase.client.RegionLocator;
45  import org.apache.hadoop.hbase.client.Result;
46  import org.apache.hadoop.hbase.client.Scan;
47  import org.apache.hadoop.hbase.client.Table;
48  import org.apache.hadoop.hbase.exceptions.IllegalArgumentIOException;
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  *
102  * The number of InputSplits(mappers) match the number of regions in a table by default.
103  * Set "hbase.mapreduce.input.mappers.per.region" to specify how many mappers per region, set
104  * this property will disable autobalance below.
105  *
106  * Set "hbase.mapreduce.input.autobalance" to enable autobalance, hbase will assign mappers based on
107  * average region size; For regions, whose size larger than average region size may assigned more mappers,
108  * and for continuous small one, they may group together to use one mapper. If actual calculated average
109  * region size is too big, it is not good to only assign 1 mapper for those large regions. Then use
110  * "hbase.mapreduce.input.average.regionsize" to set max average region size when enable "autobalanece",
111  * default was average region size is 8G.
112  * </pre>
113  *
114  */
115 @InterfaceAudience.Public
116 @InterfaceStability.Stable
117 public abstract class TableInputFormatBase
118 extends InputFormat<ImmutableBytesWritable, Result> {
119 
120   private static final Log LOG = LogFactory.getLog(TableInputFormatBase.class);
121 
122   private static final String NOT_INITIALIZED = "The input format instance has not been properly " +
123       "initialized. Ensure you call initializeTable either in your constructor or initialize " +
124       "method";
125   private static final String INITIALIZATION_ERROR = "Cannot create a record reader because of a" +
126             " previous error. Please look at the previous logs lines from" +
127             " the task's full log for more details.";
128 
129   @Deprecated
130   /** Deprecated. No effect. */
131   public static final String INPUT_AUTOBALANCE_MAXSKEWRATIO = "hbase.mapreduce.input.autobalance" +
132       ".maxskewratio";
133   @Deprecated
134   /** Deprecated. No effect. */
135   public static final String TABLE_ROW_TEXTKEY = "hbase.table.row.textkey";
136 
137   /** Specify if we enable auto-balance to set number of mappers in M/R jobs. */
138   public static final String MAPREDUCE_INPUT_AUTOBALANCE = "hbase.mapreduce.input.autobalance";
139   /** In auto-balance, we split input by ave region size, if calculated region size is too big, we can set it. */
140   public static final String MAX_AVERAGE_REGION_SIZE = "hbase.mapreduce.input.average.regionsize";
141 
142   /** Set the number of Mappers for each region, all regions have same number of Mappers */
143   public static final String NUM_MAPPERS_PER_REGION = "hbase.mapreduce.input.mappers.per.region";
144 
145   /** Holds the details for the internal scanner.
146    *
147    * @see Scan */
148   private Scan scan = null;
149   /** The {@link Admin}. */
150   private Admin admin;
151   /** The {@link Table} to scan. */
152   private Table table;
153   /** The {@link RegionLocator} of the table. */
154   private RegionLocator regionLocator;
155   /** The reader scanning the table, can be a custom one. */
156   private TableRecordReader tableRecordReader = null;
157   /** The underlying {@link Connection} of the table. */
158   private Connection connection;
159 
160 
161   /** The reverse DNS lookup cache mapping: IPAddress => HostName */
162   private HashMap<InetAddress, String> reverseDNSCacheMap =
163     new HashMap<InetAddress, String>();
164 
165   /**
166    * Builds a {@link TableRecordReader}. If no {@link TableRecordReader} was provided, uses
167    * the default.
168    *
169    * @param split  The split to work with.
170    * @param context  The current context.
171    * @return The newly created record reader.
172    * @throws IOException When creating the reader fails.
173    * @see org.apache.hadoop.mapreduce.InputFormat#createRecordReader(
174    *   org.apache.hadoop.mapreduce.InputSplit,
175    *   org.apache.hadoop.mapreduce.TaskAttemptContext)
176    */
177   @Override
178   public RecordReader<ImmutableBytesWritable, Result> createRecordReader(
179       InputSplit split, TaskAttemptContext context)
180   throws IOException {
181     // Just in case a subclass is relying on JobConfigurable magic.
182     if (table == null) {
183       initialize(context);
184     }
185     // null check in case our child overrides getTable to not throw.
186     try {
187       if (getTable() == null) {
188         // initialize() must not have been implemented in the subclass.
189         throw new IOException(INITIALIZATION_ERROR);
190       }
191     } catch (IllegalStateException exception) {
192       throw new IOException(INITIALIZATION_ERROR, exception);
193     }
194     TableSplit tSplit = (TableSplit) split;
195     LOG.info("Input split length: " + StringUtils.humanReadableInt(tSplit.getLength()) + " bytes.");
196     final TableRecordReader trr =
197         this.tableRecordReader != null ? this.tableRecordReader : new TableRecordReader();
198     Scan sc = new Scan(this.scan);
199     sc.setStartRow(tSplit.getStartRow());
200     sc.setStopRow(tSplit.getEndRow());
201     trr.setScan(sc);
202     trr.setTable(getTable());
203     return new RecordReader<ImmutableBytesWritable, Result>() {
204 
205       @Override
206       public void close() throws IOException {
207         trr.close();
208         closeTable();
209       }
210 
211       @Override
212       public ImmutableBytesWritable getCurrentKey() throws IOException, InterruptedException {
213         return trr.getCurrentKey();
214       }
215 
216       @Override
217       public Result getCurrentValue() throws IOException, InterruptedException {
218         return trr.getCurrentValue();
219       }
220 
221       @Override
222       public float getProgress() throws IOException, InterruptedException {
223         return trr.getProgress();
224       }
225 
226       @Override
227       public void initialize(InputSplit inputsplit, TaskAttemptContext context) throws IOException,
228           InterruptedException {
229         trr.initialize(inputsplit, context);
230       }
231 
232       @Override
233       public boolean nextKeyValue() throws IOException, InterruptedException {
234         return trr.nextKeyValue();
235       }
236     };
237   }
238 
239   protected Pair<byte[][],byte[][]> getStartEndKeys() throws IOException {
240     return getRegionLocator().getStartEndKeys();
241   }
242 
243   /**
244    * Calculates the splits that will serve as input for the map tasks. The
245    * number of splits matches the number of regions in a table.
246    *
247    * @param context  The current job context.
248    * @return The list of input splits.
249    * @throws IOException When creating the list of splits fails.
250    * @see org.apache.hadoop.mapreduce.InputFormat#getSplits(
251    *   org.apache.hadoop.mapreduce.JobContext)
252    */
253   @Override
254   public List<InputSplit> getSplits(JobContext context) throws IOException {
255     boolean closeOnFinish = false;
256 
257     // Just in case a subclass is relying on JobConfigurable magic.
258     if (table == null) {
259       initialize(context);
260       closeOnFinish = true;
261     }
262 
263     // null check in case our child overrides getTable to not throw.
264     try {
265       if (getTable() == null) {
266         // initialize() must not have been implemented in the subclass.
267         throw new IOException(INITIALIZATION_ERROR);
268       }
269     } catch (IllegalStateException exception) {
270       throw new IOException(INITIALIZATION_ERROR, exception);
271     }
272     try {
273       List<InputSplit> splits = oneInputSplitPerRegion();
274 
275       // set same number of mappers for each region
276       if (context.getConfiguration().get(NUM_MAPPERS_PER_REGION) != null) {
277         int nSplitsPerRegion = context.getConfiguration().getInt(NUM_MAPPERS_PER_REGION, 1);
278         List<InputSplit> res = new ArrayList<>();
279         for (int i = 0; i < splits.size(); i++) {
280           List<InputSplit> tmp = createNInputSplitsUniform(splits.get(i), nSplitsPerRegion);
281           res.addAll(tmp);
282         }
283         return res;
284       }
285 
286       //The default value of "hbase.mapreduce.input.autobalance" is false.
287       if (context.getConfiguration().getBoolean(MAPREDUCE_INPUT_AUTOBALANCE, false)) {
288         long maxAveRegionSize = context.getConfiguration()
289             .getLong(MAX_AVERAGE_REGION_SIZE, 8L*1073741824); //8GB
290         return calculateAutoBalancedSplits(splits, maxAveRegionSize);
291       }
292 
293       // return one mapper per region
294       return splits;
295     } catch (NamingException e) {
296       throw new IOException(e);
297     } finally {
298       if (closeOnFinish) {
299         closeTable();
300       }
301     }
302   }
303 
304   /**
305    * Create one InputSplit per region
306    *
307    * @return The list of InputSplit for all the regions
308    * @throws IOException
309    */
310   private List<InputSplit> oneInputSplitPerRegion() throws IOException, NamingException {
311     RegionSizeCalculator sizeCalculator =
312         new RegionSizeCalculator(getRegionLocator(), getAdmin());
313 
314     TableName tableName = getTable().getName();
315 
316     Pair<byte[][], byte[][]> keys = getStartEndKeys();
317     if (keys == null || keys.getFirst() == null ||
318         keys.getFirst().length == 0) {
319       HRegionLocation regLoc =
320           getRegionLocator().getRegionLocation(HConstants.EMPTY_BYTE_ARRAY, false);
321       if (null == regLoc) {
322         throw new IOException("Expecting at least one region.");
323       }
324       List<InputSplit> splits = new ArrayList<>(1);
325       long regionSize = sizeCalculator.getRegionSize(regLoc.getRegionInfo().getRegionName());
326       TableSplit split = new TableSplit(tableName, scan,
327           HConstants.EMPTY_BYTE_ARRAY, HConstants.EMPTY_BYTE_ARRAY, regLoc
328           .getHostnamePort().split(Addressing.HOSTNAME_PORT_SEPARATOR)[0], regionSize);
329       splits.add(split);
330       return splits;
331     }
332     List<InputSplit> splits = new ArrayList<>(keys.getFirst().length);
333     for (int i = 0; i < keys.getFirst().length; i++) {
334       if (!includeRegionInSplit(keys.getFirst()[i], keys.getSecond()[i])) {
335         continue;
336       }
337 
338       byte[] startRow = scan.getStartRow();
339       byte[] stopRow = scan.getStopRow();
340       // determine if the given start an stop key fall into the region
341       if ((startRow.length == 0 || keys.getSecond()[i].length == 0 ||
342           Bytes.compareTo(startRow, keys.getSecond()[i]) < 0) &&
343           (stopRow.length == 0 ||
344               Bytes.compareTo(stopRow, keys.getFirst()[i]) > 0)) {
345         byte[] splitStart = startRow.length == 0 ||
346             Bytes.compareTo(keys.getFirst()[i], startRow) >= 0 ?
347             keys.getFirst()[i] : startRow;
348         byte[] splitStop = (stopRow.length == 0 ||
349             Bytes.compareTo(keys.getSecond()[i], stopRow) <= 0) &&
350             keys.getSecond()[i].length > 0 ?
351             keys.getSecond()[i] : stopRow;
352 
353         HRegionLocation location = getRegionLocator().getRegionLocation(keys.getFirst()[i], false);
354         // The below InetSocketAddress creation does a name resolution.
355         InetSocketAddress isa = new InetSocketAddress(location.getHostname(), location.getPort());
356         if (isa.isUnresolved()) {
357           LOG.warn("Failed resolve " + isa);
358         }
359         InetAddress regionAddress = isa.getAddress();
360         String regionLocation;
361         regionLocation = reverseDNS(regionAddress);
362 
363         byte[] regionName = location.getRegionInfo().getRegionName();
364         String encodedRegionName = location.getRegionInfo().getEncodedName();
365         long regionSize = sizeCalculator.getRegionSize(regionName);
366         TableSplit split = new TableSplit(tableName, scan,
367             splitStart, splitStop, regionLocation, encodedRegionName, regionSize);
368         splits.add(split);
369         if (LOG.isDebugEnabled()) {
370           LOG.debug("getSplits: split -> " + i + " -> " + split);
371         }
372       }
373     }
374     return splits;
375   }
376 
377   /**
378    * Create n splits for one InputSplit, For now only support uniform distribution
379    * @param split A TableSplit corresponding to a range of rowkeys
380    * @param n     Number of ranges after splitting.  Pass 1 means no split for the range
381    *              Pass 2 if you want to split the range in two;
382    * @return A list of TableSplit, the size of the list is n
383    * @throws IllegalArgumentIOException
384    */
385   protected List<InputSplit> createNInputSplitsUniform(InputSplit split, int n)
386       throws IllegalArgumentIOException {
387     if (split == null || !(split instanceof TableSplit)) {
388       throw new IllegalArgumentIOException(
389           "InputSplit for CreateNSplitsPerRegion can not be null + "
390               + "and should be instance of TableSplit");
391     }
392     //if n < 1, then still continue using n = 1
393     n = n < 1 ? 1 : n;
394     List<InputSplit> res = new ArrayList<>(n);
395     if (n == 1) {
396       res.add(split);
397       return res;
398     }
399 
400     // Collect Region related information
401     TableSplit ts = (TableSplit) split;
402     TableName tableName = ts.getTable();
403     String regionLocation = ts.getRegionLocation();
404     String encodedRegionName = ts.getEncodedRegionName();
405     long regionSize = ts.getLength();
406     byte[] startRow = ts.getStartRow();
407     byte[] endRow = ts.getEndRow();
408 
409     // For special case: startRow or endRow is empty
410     if (startRow.length == 0 && endRow.length == 0){
411       startRow = new byte[1];
412       endRow = new byte[1];
413       startRow[0] = 0;
414       endRow[0] = -1;
415     }
416     if (startRow.length == 0 && endRow.length != 0){
417       startRow = new byte[1];
418       startRow[0] = 0;
419     }
420     if (startRow.length != 0 && endRow.length == 0){
421       endRow =new byte[startRow.length];
422       for (int k = 0; k < startRow.length; k++){
423         endRow[k] = -1;
424       }
425     }
426 
427     // Split Region into n chunks evenly
428     byte[][] splitKeys = Bytes.split(startRow, endRow, true, n-1);
429     for (int i = 0; i < splitKeys.length - 1; i++) {
430       //notice that the regionSize parameter may be not very accurate
431       TableSplit tsplit =
432           new TableSplit(tableName, scan, splitKeys[i], splitKeys[i + 1], regionLocation,
433               encodedRegionName, regionSize / n);
434       res.add(tsplit);
435     }
436     return res;
437   }
438 
439   /**
440    * Calculates the number of MapReduce input splits for the map tasks. The number of
441    * MapReduce input splits depends on the average region size.
442    * Make it 'public' for testing
443    * <p>
444    * Deprecated. Former functionality has been replaced by calculateAutoBalancedSplits and
445    * will function differently. Do not use.
446    * <p>
447    * @param list  The list of input splits before balance.
448    * @param context The current job context.
449    * @param average The average size of all regions .
450    * @return The list of input splits.
451    * @throws IOException
452    * @deprecated
453    */
454   @Deprecated
455   public List<InputSplit> calculateRebalancedSplits(List<InputSplit> list, JobContext context,
456       long average) throws IOException {
457     return calculateAutoBalancedSplits(list, average);
458   }
459 
460   /**
461    * Calculates the number of MapReduce input splits for the map tasks. The number of
462    * MapReduce input splits depends on the average region size.
463    * Make it 'public' for testing
464    *
465    * @param splits The list of input splits before balance.
466    * @param maxAverageRegionSize max Average region size for one mapper
467    * @return The list of input splits.
468    * @throws IOException When creating the list of splits fails.
469    * @see org.apache.hadoop.mapreduce.InputFormat#getSplits(
470    *org.apache.hadoop.mapreduce.JobContext)
471    */
472   public List<InputSplit> calculateAutoBalancedSplits(List<InputSplit> splits, long maxAverageRegionSize)
473       throws IOException {
474     if (splits.size() == 0) {
475       return splits;
476     }
477     List<InputSplit> resultList = new ArrayList<>();
478     long totalRegionSize = 0;
479     for (int i = 0; i < splits.size(); i++) {
480       TableSplit ts = (TableSplit) splits.get(i);
481       totalRegionSize += ts.getLength();
482     }
483     long averageRegionSize = totalRegionSize / splits.size();
484     // totalRegionSize might be overflow, and the averageRegionSize must be positive.
485     if (averageRegionSize <= 0) {
486       LOG.warn("The averageRegionSize is not positive: " + averageRegionSize + ", " +
487           "set it to Long.MAX_VALUE " + splits.size());
488       averageRegionSize = Long.MAX_VALUE / splits.size();
489     }
490     //if averageRegionSize is too big, change it to default as 8 GB,
491     if (averageRegionSize > maxAverageRegionSize) {
492       averageRegionSize = maxAverageRegionSize;
493     }
494     // if averageRegionSize is too small, we do not need to allocate more mappers for those 'large' region
495     // set default as 64M = (default hdfs block size);
496     if (averageRegionSize < 64 * 1048576) {
497       return splits;
498     }
499     for (int i = 0; i < splits.size(); i++) {
500       TableSplit ts = (TableSplit) splits.get(i);
501       TableName tableName = ts.getTable();
502       String regionLocation = ts.getRegionLocation();
503       String encodedRegionName = ts.getEncodedRegionName();
504       long regionSize = ts.getLength();
505 
506       if (regionSize >= averageRegionSize) {
507         // make this region as multiple MapReduce input split.
508         int n = (int) Math.round(Math.log(((double) regionSize) / ((double) averageRegionSize)) + 1.0);
509         List<InputSplit> temp = createNInputSplitsUniform(ts, n);
510         resultList.addAll(temp);
511       } else {
512         // if the total size of several small continuous regions less than the average region size,
513         // combine them into one MapReduce input split.
514         long totalSize = regionSize;
515         byte[] splitStartKey = ts.getStartRow();
516         byte[] splitEndKey = ts.getEndRow();
517         int j = i + 1;
518         while (j < splits.size()) {
519           TableSplit nextRegion = (TableSplit) splits.get(j);
520           long nextRegionSize = nextRegion.getLength();
521           if (totalSize + nextRegionSize <= averageRegionSize) {
522             totalSize = totalSize + nextRegionSize;
523             splitEndKey = nextRegion.getEndRow();
524             j++;
525           } else {
526             break;
527           }
528         }
529         i = j - 1;
530         TableSplit t = new TableSplit(tableName, scan, splitStartKey, splitEndKey, regionLocation,
531             encodedRegionName, totalSize);
532         resultList.add(t);
533       }
534     }
535     return resultList;
536   }
537 
538   /**
539    * Deprecated. Do not use.
540    * @param start Start key of the region
541    * @param end End key of the region
542    * @param isText It determines to use text key mode or binary key mode
543    * @return The split point in the region.
544    */
545   @Deprecated
546   public static byte[] getSplitKey(byte[] start, byte[] end, boolean isText) {
547     byte upperLimitByte;
548     byte lowerLimitByte;
549     //Use text mode or binary mode.
550     if (isText) {
551       //The range of text char set in ASCII is [32,126], the lower limit is space and the upper
552       // limit is '~'.
553       upperLimitByte = '~';
554       lowerLimitByte = ' ';
555     } else {
556       upperLimitByte = -1;
557       lowerLimitByte = 0;
558     }
559     // For special case
560     // Example 1 : startkey=null, endkey="hhhqqqwww", splitKey="h"
561     // Example 2 (text key mode): startKey="ffffaaa", endKey=null, splitkey="f~~~~~~"
562     if (start.length == 0 && end.length == 0){
563       return new byte[]{(byte) ((lowerLimitByte + upperLimitByte) / 2)};
564     }
565     if (start.length == 0 && end.length != 0){
566       return new byte[]{ end[0] };
567     }
568     if (start.length != 0 && end.length == 0){
569       byte[] result =new byte[start.length];
570       result[0]=start[0];
571       return result;
572     }
573     return Bytes.split(start, end, false, 1)[1];
574   }
575 
576   /**
577    * @deprecated mistakenly made public in 0.98.7. scope will change to package-private
578    */
579   @Deprecated
580   public String reverseDNS(InetAddress ipAddress) throws NamingException, UnknownHostException {
581     String hostName = this.reverseDNSCacheMap.get(ipAddress);
582     if (hostName == null) {
583       String ipAddressString = null;
584       try {
585         ipAddressString = DNS.reverseDns(ipAddress, null);
586       } catch (Exception e) {
587         // We can use InetAddress in case the jndi failed to pull up the reverse DNS entry from the
588         // name service. Also, in case of ipv6, we need to use the InetAddress since resolving
589         // reverse DNS using jndi doesn't work well with ipv6 addresses.
590         ipAddressString = InetAddress.getByName(ipAddress.getHostAddress()).getHostName();
591       }
592       if (ipAddressString == null) throw new UnknownHostException("No host found for " + ipAddress);
593       hostName = Strings.domainNamePointerToHostName(ipAddressString);
594       this.reverseDNSCacheMap.put(ipAddress, hostName);
595     }
596     return hostName;
597   }
598 
599   /**
600    *
601    *
602    * Test if the given region is to be included in the InputSplit while splitting
603    * the regions of a table.
604    * <p>
605    * This optimization is effective when there is a specific reasoning to exclude an entire region from the M-R job,
606    * (and hence, not contributing to the InputSplit), given the start and end keys of the same. <br>
607    * Useful when we need to remember the last-processed top record and revisit the [last, current) interval for M-R processing,
608    * continuously. In addition to reducing InputSplits, reduces the load on the region server as
609    * well, due to the ordering of the keys.
610    * <br>
611    * <br>
612    * Note: It is possible that <code>endKey.length() == 0 </code> , for the last (recent) region.
613    * <br>
614    * Override this method, if you want to bulk exclude regions altogether from M-R.
615    * By default, no region is excluded( i.e. all regions are included).
616    *
617    *
618    * @param startKey Start key of the region
619    * @param endKey End key of the region
620    * @return true, if this region needs to be included as part of the input (default).
621    *
622    */
623   protected boolean includeRegionInSplit(final byte[] startKey, final byte [] endKey) {
624     return true;
625   }
626 
627   /**
628    * Allows subclasses to get the {@link HTable}.
629    *
630    * @deprecated use {@link #getTable()}
631    */
632   @Deprecated
633   protected HTable getHTable() {
634     return (HTable) this.getTable();
635   }
636 
637   /**
638    * Allows subclasses to get the {@link RegionLocator}.
639    */
640   protected RegionLocator getRegionLocator() {
641     if (regionLocator == null) {
642       throw new IllegalStateException(NOT_INITIALIZED);
643     }
644     return regionLocator;
645   }
646 
647   /**
648    * Allows subclasses to get the {@link Table}.
649    */
650   protected Table getTable() {
651     if (table == null) {
652       throw new IllegalStateException(NOT_INITIALIZED);
653     }
654     return table;
655   }
656 
657   /**
658    * Allows subclasses to get the {@link Admin}.
659    */
660   protected Admin getAdmin() {
661     if (admin == null) {
662       throw new IllegalStateException(NOT_INITIALIZED);
663     }
664     return admin;
665   }
666 
667   /**
668    * Allows subclasses to set the {@link HTable}.
669    *
670    * Will attempt to reuse the underlying Connection for our own needs, including
671    * retreiving an Admin interface to the HBase cluster.
672    *
673    * @param table  The table to get the data from.
674    * @throws IOException
675    * @deprecated Use {@link #initializeTable(Connection, TableName)} instead.
676    */
677   @Deprecated
678   protected void setHTable(HTable table) throws IOException {
679     this.table = table;
680     this.connection = table.getConnection();
681     try {
682       this.regionLocator = table.getRegionLocator();
683       this.admin = this.connection.getAdmin();
684     } catch (NeedUnmanagedConnectionException exception) {
685       LOG.warn("You are using an HTable instance that relies on an HBase-managed Connection. " +
686           "This is usually due to directly creating an HTable, which is deprecated. Instead, you " +
687           "should create a Connection object and then request a Table instance from it. If you " +
688           "don't need the Table instance for your own use, you should instead use the " +
689           "TableInputFormatBase.initalizeTable method directly.");
690       LOG.info("Creating an additional unmanaged connection because user provided one can't be " +
691           "used for administrative actions. We'll close it when we close out the table.");
692       LOG.debug("Details about our failure to request an administrative interface.", exception);
693       // Do we need a "copy the settings from this Connection" method? are things like the User
694       // properly maintained by just looking again at the Configuration?
695       this.connection = ConnectionFactory.createConnection(this.connection.getConfiguration());
696       this.regionLocator = this.connection.getRegionLocator(table.getName());
697       this.admin = this.connection.getAdmin();
698     }
699   }
700 
701   /**
702    * Allows subclasses to initialize the table information.
703    *
704    * @param connection  The {@link Connection} to the HBase cluster. MUST be unmanaged. We will close.
705    * @param tableName  The {@link TableName} of the table to process.
706    * @throws IOException
707    */
708   protected void initializeTable(Connection connection, TableName tableName) throws IOException {
709     if (this.table != null || this.connection != null) {
710       LOG.warn("initializeTable called multiple times. Overwriting connection and table " +
711           "reference; TableInputFormatBase will not close these old references when done.");
712     }
713     this.table = connection.getTable(tableName);
714     this.regionLocator = connection.getRegionLocator(tableName);
715     this.admin = connection.getAdmin();
716     this.connection = connection;
717   }
718 
719   /**
720    * Gets the scan defining the actual details like columns etc.
721    *
722    * @return The internal scan instance.
723    */
724   public Scan getScan() {
725     if (this.scan == null) this.scan = new Scan();
726     return scan;
727   }
728 
729   /**
730    * Sets the scan defining the actual details like columns etc.
731    *
732    * @param scan  The scan to set.
733    */
734   public void setScan(Scan scan) {
735     this.scan = scan;
736   }
737 
738   /**
739    * Allows subclasses to set the {@link TableRecordReader}.
740    *
741    * @param tableRecordReader A different {@link TableRecordReader}
742    *   implementation.
743    */
744   protected void setTableRecordReader(TableRecordReader tableRecordReader) {
745     this.tableRecordReader = tableRecordReader;
746   }
747 
748   /**
749    * Handle subclass specific set up.
750    * Each of the entry points used by the MapReduce framework,
751    * {@link #createRecordReader(InputSplit, TaskAttemptContext)} and {@link #getSplits(JobContext)},
752    * will call {@link #initialize(JobContext)} as a convenient centralized location to handle
753    * retrieving the necessary configuration information and calling
754    * {@link #initializeTable(Connection, TableName)}.
755    *
756    * Subclasses should implement their initialize call such that it is safe to call multiple times.
757    * The current TableInputFormatBase implementation relies on a non-null table reference to decide
758    * if an initialize call is needed, but this behavior may change in the future. In particular,
759    * it is critical that initializeTable not be called multiple times since this will leak
760    * Connection instances.
761    *
762    */
763   protected void initialize(JobContext context) throws IOException {
764   }
765 
766   /**
767    * Close the Table and related objects that were initialized via
768    * {@link #initializeTable(Connection, TableName)}.
769    *
770    * @throws IOException
771    */
772   protected void closeTable() throws IOException {
773     close(admin, table, regionLocator, connection);
774     admin = null;
775     table = null;
776     regionLocator = null;
777     connection = null;
778   }
779 
780   private void close(Closeable... closables) throws IOException {
781     for (Closeable c : closables) {
782       if(c != null) { c.close(); }
783     }
784   }
785 
786 }