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