View Javadoc

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