View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
17   */
18  package org.apache.hadoop.hbase.mapreduce;
19  
20  import java.io.IOException;
21  import java.lang.reflect.Method;
22  import java.util.Map;
23  
24  import org.apache.commons.logging.Log;
25  import org.apache.commons.logging.LogFactory;
26  import org.apache.hadoop.hbase.classification.InterfaceAudience;
27  import org.apache.hadoop.hbase.classification.InterfaceStability;
28  import org.apache.hadoop.conf.Configuration;
29  import org.apache.hadoop.hbase.client.Result;
30  import org.apache.hadoop.hbase.client.ResultScanner;
31  import org.apache.hadoop.hbase.client.Scan;
32  import org.apache.hadoop.hbase.client.ScannerCallable;
33  import org.apache.hadoop.hbase.client.Table;
34  import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
35  import org.apache.hadoop.hbase.DoNotRetryIOException;
36  import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
37  import org.apache.hadoop.hbase.util.Bytes;
38  import org.apache.hadoop.mapreduce.Counter;
39  import org.apache.hadoop.mapreduce.InputSplit;
40  import org.apache.hadoop.mapreduce.TaskAttemptContext;
41  import org.apache.hadoop.util.StringUtils;
42  
43  import com.google.common.annotations.VisibleForTesting;
44  
45  /**
46   * Iterate over an HBase table data, return (ImmutableBytesWritable, Result)
47   * pairs.
48   */
49  @InterfaceAudience.Public
50  @InterfaceStability.Stable
51  public class TableRecordReaderImpl {
52    public static final String LOG_PER_ROW_COUNT
53      = "hbase.mapreduce.log.scanner.rowcount";
54  
55    private static final Log LOG = LogFactory.getLog(TableRecordReaderImpl.class);
56  
57    // HBASE_COUNTER_GROUP_NAME is the name of mapreduce counter group for HBase
58    @VisibleForTesting
59    static final String HBASE_COUNTER_GROUP_NAME = "HBase Counters";
60    private ResultScanner scanner = null;
61    private Scan scan = null;
62    private Scan currentScan = null;
63    private Table htable = null;
64    private byte[] lastSuccessfulRow = null;
65    private ImmutableBytesWritable key = null;
66    private Result value = null;
67    private TaskAttemptContext context = null;
68    private Method getCounter = null;
69    private long numRestarts = 0;
70    private long numStale = 0;
71    private long timestamp;
72    private int rowcount;
73    private boolean logScannerActivity = false;
74    private int logPerRowCount = 100;
75  
76    /**
77     * Restart from survivable exceptions by creating a new scanner.
78     *
79     * @param firstRow  The first row to start at.
80     * @throws IOException When restarting fails.
81     */
82    public void restart(byte[] firstRow) throws IOException {
83      currentScan = new Scan(scan);
84      currentScan.setStartRow(firstRow);
85      currentScan.setScanMetricsEnabled(true);
86      if (this.scanner != null) {
87        if (logScannerActivity) {
88          LOG.info("Closing the previously opened scanner object.");
89        }
90        this.scanner.close();
91      }
92      this.scanner = this.htable.getScanner(currentScan);
93      if (logScannerActivity) {
94        LOG.info("Current scan=" + currentScan.toString());
95        timestamp = System.currentTimeMillis();
96        rowcount = 0;
97      }
98    }
99  
100   /**
101    * In new mapreduce APIs, TaskAttemptContext has two getCounter methods
102    * Check if getCounter(String, String) method is available.
103    * @return The getCounter method or null if not available.
104    * @throws IOException
105    */
106   protected static Method retrieveGetCounterWithStringsParams(TaskAttemptContext context)
107   throws IOException {
108     Method m = null;
109     try {
110       m = context.getClass().getMethod("getCounter",
111         new Class [] {String.class, String.class});
112     } catch (SecurityException e) {
113       throw new IOException("Failed test for getCounter", e);
114     } catch (NoSuchMethodException e) {
115       // Ignore
116     }
117     return m;
118   }
119 
120   /**
121    * Sets the HBase table.
122    *
123    * @param htable  The {@link org.apache.hadoop.hbase.HTableDescriptor} to scan.
124    */
125   public void setHTable(Table htable) {
126     Configuration conf = htable.getConfiguration();
127     logScannerActivity = conf.getBoolean(
128       ScannerCallable.LOG_SCANNER_ACTIVITY, false);
129     logPerRowCount = conf.getInt(LOG_PER_ROW_COUNT, 100);
130     this.htable = htable;
131   }
132 
133   /**
134    * Sets the scan defining the actual details like columns etc.
135    *
136    * @param scan  The scan to set.
137    */
138   public void setScan(Scan scan) {
139     this.scan = scan;
140   }
141 
142   /**
143    * Build the scanner. Not done in constructor to allow for extension.
144    *
145    * @throws IOException
146    * @throws InterruptedException
147    */
148   public void initialize(InputSplit inputsplit,
149       TaskAttemptContext context) throws IOException,
150       InterruptedException {
151     if (context != null) {
152       this.context = context;
153       getCounter = retrieveGetCounterWithStringsParams(context);
154     }
155     restart(scan.getStartRow());
156   }
157 
158   /**
159    * Closes the split.
160    *
161    *
162    */
163   public void close() {
164     this.scanner.close();
165     try {
166       this.htable.close();
167     } catch (IOException ioe) {
168       LOG.warn("Error closing table", ioe);
169     }
170   }
171 
172   /**
173    * Returns the current key.
174    *
175    * @return The current key.
176    * @throws IOException
177    * @throws InterruptedException When the job is aborted.
178    */
179   public ImmutableBytesWritable getCurrentKey() throws IOException,
180       InterruptedException {
181     return key;
182   }
183 
184   /**
185    * Returns the current value.
186    *
187    * @return The current value.
188    * @throws IOException When the value is faulty.
189    * @throws InterruptedException When the job is aborted.
190    */
191   public Result getCurrentValue() throws IOException, InterruptedException {
192     return value;
193   }
194 
195 
196   /**
197    * Positions the record reader to the next record.
198    *
199    * @return <code>true</code> if there was another record.
200    * @throws IOException When reading the record failed.
201    * @throws InterruptedException When the job was aborted.
202    */
203   public boolean nextKeyValue() throws IOException, InterruptedException {
204     if (key == null) key = new ImmutableBytesWritable();
205     if (value == null) value = new Result();
206     try {
207       try {
208         value = this.scanner.next();
209         if (value != null && value.isStale()) numStale++;
210         if (logScannerActivity) {
211           rowcount ++;
212           if (rowcount >= logPerRowCount) {
213             long now = System.currentTimeMillis();
214             LOG.info("Mapper took " + (now-timestamp)
215               + "ms to process " + rowcount + " rows");
216             timestamp = now;
217             rowcount = 0;
218           }
219         }
220       } catch (IOException e) {
221         // do not retry if the exception tells us not to do so
222         if (e instanceof DoNotRetryIOException) {
223           throw e;
224         }
225         // try to handle all other IOExceptions by restarting
226         // the scanner, if the second call fails, it will be rethrown
227         LOG.info("recovered from " + StringUtils.stringifyException(e));
228         if (lastSuccessfulRow == null) {
229           LOG.warn("We are restarting the first next() invocation," +
230               " if your mapper has restarted a few other times like this" +
231               " then you should consider killing this job and investigate" +
232               " why it's taking so long.");
233         }
234         if (lastSuccessfulRow == null) {
235           restart(scan.getStartRow());
236         } else {
237           restart(lastSuccessfulRow);
238           scanner.next();    // skip presumed already mapped row
239         }
240         value = scanner.next();
241         if (value != null && value.isStale()) numStale++;
242         numRestarts++;
243       }
244       if (value != null && value.size() > 0) {
245         key.set(value.getRow());
246         lastSuccessfulRow = key.get();
247         return true;
248       }
249 
250       updateCounters();
251       return false;
252     } catch (IOException ioe) {
253       if (logScannerActivity) {
254         long now = System.currentTimeMillis();
255         LOG.info("Mapper took " + (now-timestamp)
256           + "ms to process " + rowcount + " rows");
257         LOG.info(ioe);
258         String lastRow = lastSuccessfulRow == null ?
259           "null" : Bytes.toStringBinary(lastSuccessfulRow);
260         LOG.info("lastSuccessfulRow=" + lastRow);
261       }
262       throw ioe;
263     }
264   }
265 
266   /**
267    * If hbase runs on new version of mapreduce, RecordReader has access to
268    * counters thus can update counters based on scanMetrics.
269    * If hbase runs on old version of mapreduce, it won't be able to get
270    * access to counters and TableRecorderReader can't update counter values.
271    * @throws IOException
272    */
273   private void updateCounters() throws IOException {
274     ScanMetrics scanMetrics = currentScan.getScanMetrics();
275     if (scanMetrics == null) {
276       return;
277     }
278 
279     updateCounters(scanMetrics, numRestarts, getCounter, context, numStale);
280   }
281 
282   protected static void updateCounters(ScanMetrics scanMetrics, long numScannerRestarts,
283       Method getCounter, TaskAttemptContext context, long numStale) {
284     // we can get access to counters only if hbase uses new mapreduce APIs
285     if (getCounter == null) {
286       return;
287     }
288 
289     try {
290       for (Map.Entry<String, Long> entry:scanMetrics.getMetricsMap().entrySet()) {
291         Counter ct = (Counter)getCounter.invoke(context,
292             HBASE_COUNTER_GROUP_NAME, entry.getKey());
293 
294         ct.increment(entry.getValue());
295       }
296       ((Counter) getCounter.invoke(context, HBASE_COUNTER_GROUP_NAME,
297           "NUM_SCANNER_RESTARTS")).increment(numScannerRestarts);
298       ((Counter) getCounter.invoke(context, HBASE_COUNTER_GROUP_NAME,
299           "NUM_SCAN_RESULTS_STALE")).increment(numStale);
300     } catch (Exception e) {
301       LOG.debug("can't update counter." + StringUtils.stringifyException(e));
302     }
303   }
304 
305   /**
306    * The current progress of the record reader through its data.
307    *
308    * @return A number between 0.0 and 1.0, the fraction of the data read.
309    */
310   public float getProgress() {
311     // Depends on the total number of tuples
312     return 0;
313   }
314 
315 }