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