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 com.google.protobuf.InvalidProtocolBufferException;
22  import com.yammer.metrics.core.MetricsRegistry;
23  import org.apache.commons.logging.Log;
24  import org.apache.commons.logging.LogFactory;
25  import org.apache.hadoop.conf.Configuration;
26  import org.apache.hadoop.fs.FileSystem;
27  import org.apache.hadoop.fs.Path;
28  import org.apache.hadoop.hbase.HBaseConfiguration;
29  import org.apache.hadoop.hbase.HConstants;
30  import org.apache.hadoop.hbase.MetaTableAccessor;
31  import org.apache.hadoop.hbase.TableName;
32  import org.apache.hadoop.hbase.classification.InterfaceAudience;
33  import org.apache.hadoop.hbase.classification.InterfaceStability;
34  import org.apache.hadoop.hbase.client.Connection;
35  import org.apache.hadoop.hbase.client.ConnectionFactory;
36  import org.apache.hadoop.hbase.client.Put;
37  import org.apache.hadoop.hbase.client.Scan;
38  import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
39  import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
40  import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
41  import org.apache.hadoop.hbase.security.User;
42  import org.apache.hadoop.hbase.security.UserProvider;
43  import org.apache.hadoop.hbase.security.token.TokenUtil;
44  import org.apache.hadoop.hbase.util.Base64;
45  import org.apache.hadoop.hbase.util.Bytes;
46  import org.apache.hadoop.hbase.zookeeper.ZKUtil;
47  import org.apache.hadoop.io.Writable;
48  import org.apache.hadoop.mapreduce.InputFormat;
49  import org.apache.hadoop.mapreduce.Job;
50  import org.apache.hadoop.util.StringUtils;
51  import com.google.protobuf.InvalidProtocolBufferException;
52  
53  import java.io.File;
54  import java.io.IOException;
55  import java.net.URL;
56  import java.net.URLDecoder;
57  import java.util.ArrayList;
58  import java.util.Collection;
59  import java.util.Enumeration;
60  import java.util.HashMap;
61  import java.util.HashSet;
62  import java.util.List;
63  import java.util.Map;
64  import java.util.Set;
65  import java.util.zip.ZipEntry;
66  import java.util.zip.ZipFile;
67  
68  /**
69   * Utility for {@link TableMapper} and {@link TableReducer}
70   */
71  @SuppressWarnings({ "rawtypes", "unchecked" })
72  @InterfaceAudience.Public
73  @InterfaceStability.Stable
74  public class TableMapReduceUtil {
75    private static final Log LOG = LogFactory.getLog(TableMapReduceUtil.class);
76  
77    /**
78     * Use this before submitting a TableMap job. It will appropriately set up
79     * the job.
80     *
81     * @param table  The table name to read from.
82     * @param scan  The scan instance with the columns, time range etc.
83     * @param mapper  The mapper class to use.
84     * @param outputKeyClass  The class of the output key.
85     * @param outputValueClass  The class of the output value.
86     * @param job  The current job to adjust.  Make sure the passed job is
87     * carrying all necessary HBase configuration.
88     * @throws IOException When setting up the details fails.
89     */
90    public static void initTableMapperJob(String table, Scan scan,
91        Class<? extends TableMapper> mapper,
92        Class<?> outputKeyClass,
93        Class<?> outputValueClass, Job job)
94    throws IOException {
95      initTableMapperJob(table, scan, mapper, outputKeyClass, outputValueClass,
96          job, true);
97    }
98  
99  
100   /**
101    * Use this before submitting a TableMap job. It will appropriately set up
102    * the job.
103    *
104    * @param table  The table name to read from.
105    * @param scan  The scan instance with the columns, time range etc.
106    * @param mapper  The mapper class to use.
107    * @param outputKeyClass  The class of the output key.
108    * @param outputValueClass  The class of the output value.
109    * @param job  The current job to adjust.  Make sure the passed job is
110    * carrying all necessary HBase configuration.
111    * @throws IOException When setting up the details fails.
112    */
113   public static void initTableMapperJob(TableName table,
114       Scan scan,
115       Class<? extends TableMapper> mapper,
116       Class<?> outputKeyClass,
117       Class<?> outputValueClass,
118       Job job) throws IOException {
119     initTableMapperJob(table.getNameAsString(),
120         scan,
121         mapper,
122         outputKeyClass,
123         outputValueClass,
124         job,
125         true);
126   }
127 
128   /**
129    * Use this before submitting a TableMap job. It will appropriately set up
130    * the job.
131    *
132    * @param table Binary representation of the table name to read from.
133    * @param scan  The scan instance with the columns, time range etc.
134    * @param mapper  The mapper class to use.
135    * @param outputKeyClass  The class of the output key.
136    * @param outputValueClass  The class of the output value.
137    * @param job  The current job to adjust.  Make sure the passed job is
138    * carrying all necessary HBase configuration.
139    * @throws IOException When setting up the details fails.
140    */
141    public static void initTableMapperJob(byte[] table, Scan scan,
142       Class<? extends TableMapper> mapper,
143       Class<?> outputKeyClass,
144       Class<?> outputValueClass, Job job)
145   throws IOException {
146       initTableMapperJob(Bytes.toString(table), scan, mapper, outputKeyClass, outputValueClass,
147               job, true);
148   }
149 
150    /**
151     * Use this before submitting a TableMap job. It will appropriately set up
152     * the job.
153     *
154     * @param table  The table name to read from.
155     * @param scan  The scan instance with the columns, time range etc.
156     * @param mapper  The mapper class to use.
157     * @param outputKeyClass  The class of the output key.
158     * @param outputValueClass  The class of the output value.
159     * @param job  The current job to adjust.  Make sure the passed job is
160     * carrying all necessary HBase configuration.
161     * @param addDependencyJars upload HBase jars and jars for any of the configured
162     *           job classes via the distributed cache (tmpjars).
163     * @throws IOException When setting up the details fails.
164     */
165    public static void initTableMapperJob(String table, Scan scan,
166        Class<? extends TableMapper> mapper,
167        Class<?> outputKeyClass,
168        Class<?> outputValueClass, Job job,
169        boolean addDependencyJars, Class<? extends InputFormat> inputFormatClass)
170    throws IOException {
171      initTableMapperJob(table, scan, mapper, outputKeyClass, outputValueClass, job,
172          addDependencyJars, true, inputFormatClass);
173    }
174 
175 
176   /**
177    * Use this before submitting a TableMap job. It will appropriately set up
178    * the job.
179    *
180    * @param table  The table name to read from.
181    * @param scan  The scan instance with the columns, time range etc.
182    * @param mapper  The mapper class to use.
183    * @param outputKeyClass  The class of the output key.
184    * @param outputValueClass  The class of the output value.
185    * @param job  The current job to adjust.  Make sure the passed job is
186    * carrying all necessary HBase configuration.
187    * @param addDependencyJars upload HBase jars and jars for any of the configured
188    *           job classes via the distributed cache (tmpjars).
189    * @param initCredentials whether to initialize hbase auth credentials for the job
190    * @param inputFormatClass the input format
191    * @throws IOException When setting up the details fails.
192    */
193   public static void initTableMapperJob(String table, Scan scan,
194       Class<? extends TableMapper> mapper,
195       Class<?> outputKeyClass,
196       Class<?> outputValueClass, Job job,
197       boolean addDependencyJars, boolean initCredentials,
198       Class<? extends InputFormat> inputFormatClass)
199   throws IOException {
200     job.setInputFormatClass(inputFormatClass);
201     if (outputValueClass != null) job.setMapOutputValueClass(outputValueClass);
202     if (outputKeyClass != null) job.setMapOutputKeyClass(outputKeyClass);
203     job.setMapperClass(mapper);
204     if (Put.class.equals(outputValueClass)) {
205       job.setCombinerClass(PutCombiner.class);
206     }
207     Configuration conf = job.getConfiguration();
208     HBaseConfiguration.merge(conf, HBaseConfiguration.create(conf));
209     conf.set(TableInputFormat.INPUT_TABLE, table);
210     conf.set(TableInputFormat.SCAN, convertScanToString(scan));
211     conf.setStrings("io.serializations", conf.get("io.serializations"),
212         MutationSerialization.class.getName(), ResultSerialization.class.getName(),
213         KeyValueSerialization.class.getName());
214     if (addDependencyJars) {
215       addDependencyJars(job);
216     }
217     if (initCredentials) {
218       initCredentials(job);
219     }
220   }
221 
222   /**
223    * Use this before submitting a TableMap job. It will appropriately set up
224    * the job.
225    *
226    * @param table Binary representation of the table name to read from.
227    * @param scan  The scan instance with the columns, time range etc.
228    * @param mapper  The mapper class to use.
229    * @param outputKeyClass  The class of the output key.
230    * @param outputValueClass  The class of the output value.
231    * @param job  The current job to adjust.  Make sure the passed job is
232    * carrying all necessary HBase configuration.
233    * @param addDependencyJars upload HBase jars and jars for any of the configured
234    *           job classes via the distributed cache (tmpjars).
235    * @param inputFormatClass The class of the input format
236    * @throws IOException When setting up the details fails.
237    */
238   public static void initTableMapperJob(byte[] table, Scan scan,
239       Class<? extends TableMapper> mapper,
240       Class<?> outputKeyClass,
241       Class<?> outputValueClass, Job job,
242       boolean addDependencyJars, Class<? extends InputFormat> inputFormatClass)
243   throws IOException {
244       initTableMapperJob(Bytes.toString(table), scan, mapper, outputKeyClass,
245               outputValueClass, job, addDependencyJars, inputFormatClass);
246   }
247 
248   /**
249    * Use this before submitting a TableMap job. It will appropriately set up
250    * the job.
251    *
252    * @param table Binary representation of the table name to read from.
253    * @param scan  The scan instance with the columns, time range etc.
254    * @param mapper  The mapper class to use.
255    * @param outputKeyClass  The class of the output key.
256    * @param outputValueClass  The class of the output value.
257    * @param job  The current job to adjust.  Make sure the passed job is
258    * carrying all necessary HBase configuration.
259    * @param addDependencyJars upload HBase jars and jars for any of the configured
260    *           job classes via the distributed cache (tmpjars).
261    * @throws IOException When setting up the details fails.
262    */
263   public static void initTableMapperJob(byte[] table, Scan scan,
264       Class<? extends TableMapper> mapper,
265       Class<?> outputKeyClass,
266       Class<?> outputValueClass, Job job,
267       boolean addDependencyJars)
268   throws IOException {
269       initTableMapperJob(Bytes.toString(table), scan, mapper, outputKeyClass,
270               outputValueClass, job, addDependencyJars, TableInputFormat.class);
271   }
272 
273   /**
274    * Use this before submitting a TableMap job. It will appropriately set up
275    * the job.
276    *
277    * @param table The table name to read from.
278    * @param scan  The scan instance with the columns, time range etc.
279    * @param mapper  The mapper class to use.
280    * @param outputKeyClass  The class of the output key.
281    * @param outputValueClass  The class of the output value.
282    * @param job  The current job to adjust.  Make sure the passed job is
283    * carrying all necessary HBase configuration.
284    * @param addDependencyJars upload HBase jars and jars for any of the configured
285    *           job classes via the distributed cache (tmpjars).
286    * @throws IOException When setting up the details fails.
287    */
288   public static void initTableMapperJob(String table, Scan scan,
289       Class<? extends TableMapper> mapper,
290       Class<?> outputKeyClass,
291       Class<?> outputValueClass, Job job,
292       boolean addDependencyJars)
293   throws IOException {
294       initTableMapperJob(table, scan, mapper, outputKeyClass,
295               outputValueClass, job, addDependencyJars, TableInputFormat.class);
296   }
297 
298   /**
299    * Enable a basic on-heap cache for these jobs. Any BlockCache implementation based on
300    * direct memory will likely cause the map tasks to OOM when opening the region. This
301    * is done here instead of in TableSnapshotRegionRecordReader in case an advanced user
302    * wants to override this behavior in their job.
303    */
304   public static void resetCacheConfig(Configuration conf) {
305     conf.setFloat(
306       HConstants.HFILE_BLOCK_CACHE_SIZE_KEY, HConstants.HFILE_BLOCK_CACHE_SIZE_DEFAULT);
307     conf.setFloat(HConstants.BUCKET_CACHE_SIZE_KEY, 0f);
308     conf.unset(HConstants.BUCKET_CACHE_IOENGINE_KEY);
309   }
310 
311   /**
312    * Sets up the job for reading from one or more table snapshots, with one or more scans
313    * per snapshot.
314    * It bypasses hbase servers and read directly from snapshot files.
315    *
316    * @param snapshotScans     map of snapshot name to scans on that snapshot.
317    * @param mapper            The mapper class to use.
318    * @param outputKeyClass    The class of the output key.
319    * @param outputValueClass  The class of the output value.
320    * @param job               The current job to adjust.  Make sure the passed job is
321    *                          carrying all necessary HBase configuration.
322    * @param addDependencyJars upload HBase jars and jars for any of the configured
323    *                          job classes via the distributed cache (tmpjars).
324    */
325   public static void initMultiTableSnapshotMapperJob(Map<String, Collection<Scan>> snapshotScans,
326       Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass,
327       Job job, boolean addDependencyJars, Path tmpRestoreDir) throws IOException {
328     MultiTableSnapshotInputFormat.setInput(job.getConfiguration(), snapshotScans, tmpRestoreDir);
329 
330     job.setInputFormatClass(MultiTableSnapshotInputFormat.class);
331     if (outputValueClass != null) {
332       job.setMapOutputValueClass(outputValueClass);
333     }
334     if (outputKeyClass != null) {
335       job.setMapOutputKeyClass(outputKeyClass);
336     }
337     job.setMapperClass(mapper);
338     Configuration conf = job.getConfiguration();
339     HBaseConfiguration.merge(conf, HBaseConfiguration.create(conf));
340 
341     if (addDependencyJars) {
342       addDependencyJars(job);
343       addDependencyJars(job.getConfiguration(), MetricsRegistry.class);
344     }
345 
346     resetCacheConfig(job.getConfiguration());
347   }
348 
349   /**
350    * Sets up the job for reading from a table snapshot. It bypasses hbase servers
351    * and read directly from snapshot files.
352    *
353    * @param snapshotName The name of the snapshot (of a table) to read from.
354    * @param scan  The scan instance with the columns, time range etc.
355    * @param mapper  The mapper class to use.
356    * @param outputKeyClass  The class of the output key.
357    * @param outputValueClass  The class of the output value.
358    * @param job  The current job to adjust.  Make sure the passed job is
359    * carrying all necessary HBase configuration.
360    * @param addDependencyJars upload HBase jars and jars for any of the configured
361    *           job classes via the distributed cache (tmpjars).
362    *
363    * @param tmpRestoreDir a temporary directory to copy the snapshot files into. Current user should
364    * have write permissions to this directory, and this should not be a subdirectory of rootdir.
365    * After the job is finished, restore directory can be deleted.
366    * @throws IOException When setting up the details fails.
367    * @see TableSnapshotInputFormat
368    */
369   public static void initTableSnapshotMapperJob(String snapshotName, Scan scan,
370       Class<? extends TableMapper> mapper,
371       Class<?> outputKeyClass,
372       Class<?> outputValueClass, Job job,
373       boolean addDependencyJars, Path tmpRestoreDir)
374   throws IOException {
375     TableSnapshotInputFormat.setInput(job, snapshotName, tmpRestoreDir);
376     initTableMapperJob(snapshotName, scan, mapper, outputKeyClass,
377         outputValueClass, job, addDependencyJars, false, TableSnapshotInputFormat.class);
378     resetCacheConfig(job.getConfiguration());
379   }
380 
381   /**
382    * Use this before submitting a Multi TableMap job. It will appropriately set
383    * up the job.
384    *
385    * @param scans The list of {@link Scan} objects to read from.
386    * @param mapper The mapper class to use.
387    * @param outputKeyClass The class of the output key.
388    * @param outputValueClass The class of the output value.
389    * @param job The current job to adjust. Make sure the passed job is carrying
390    *          all necessary HBase configuration.
391    * @throws IOException When setting up the details fails.
392    */
393   public static void initTableMapperJob(List<Scan> scans,
394       Class<? extends TableMapper> mapper,
395       Class<?> outputKeyClass,
396       Class<?> outputValueClass, Job job) throws IOException {
397     initTableMapperJob(scans, mapper, outputKeyClass, outputValueClass, job,
398         true);
399   }
400 
401   /**
402    * Use this before submitting a Multi TableMap job. It will appropriately set
403    * up the job.
404    *
405    * @param scans The list of {@link Scan} objects to read from.
406    * @param mapper The mapper class to use.
407    * @param outputKeyClass The class of the output key.
408    * @param outputValueClass The class of the output value.
409    * @param job The current job to adjust. Make sure the passed job is carrying
410    *          all necessary HBase configuration.
411    * @param addDependencyJars upload HBase jars and jars for any of the
412    *          configured job classes via the distributed cache (tmpjars).
413    * @throws IOException When setting up the details fails.
414    */
415   public static void initTableMapperJob(List<Scan> scans,
416       Class<? extends TableMapper> mapper,
417       Class<?> outputKeyClass,
418       Class<?> outputValueClass, Job job,
419       boolean addDependencyJars) throws IOException {
420     initTableMapperJob(scans, mapper, outputKeyClass, outputValueClass, job,
421       addDependencyJars, true);
422   }
423 
424   /**
425    * Use this before submitting a Multi TableMap job. It will appropriately set
426    * up the job.
427    *
428    * @param scans The list of {@link Scan} objects to read from.
429    * @param mapper The mapper class to use.
430    * @param outputKeyClass The class of the output key.
431    * @param outputValueClass The class of the output value.
432    * @param job The current job to adjust. Make sure the passed job is carrying
433    *          all necessary HBase configuration.
434    * @param addDependencyJars upload HBase jars and jars for any of the
435    *          configured job classes via the distributed cache (tmpjars).
436    * @param initCredentials whether to initialize hbase auth credentials for the job
437    * @throws IOException When setting up the details fails.
438    */
439   public static void initTableMapperJob(List<Scan> scans,
440       Class<? extends TableMapper> mapper,
441       Class<?> outputKeyClass,
442       Class<?> outputValueClass, Job job,
443       boolean addDependencyJars,
444       boolean initCredentials) throws IOException {
445     job.setInputFormatClass(MultiTableInputFormat.class);
446     if (outputValueClass != null) {
447       job.setMapOutputValueClass(outputValueClass);
448     }
449     if (outputKeyClass != null) {
450       job.setMapOutputKeyClass(outputKeyClass);
451     }
452     job.setMapperClass(mapper);
453     Configuration conf = job.getConfiguration();
454     HBaseConfiguration.merge(conf, HBaseConfiguration.create(conf));
455     List<String> scanStrings = new ArrayList<String>();
456 
457     for (Scan scan : scans) {
458       scanStrings.add(convertScanToString(scan));
459     }
460     job.getConfiguration().setStrings(MultiTableInputFormat.SCANS,
461       scanStrings.toArray(new String[scanStrings.size()]));
462 
463     if (addDependencyJars) {
464       addDependencyJars(job);
465     }
466 
467     if (initCredentials) {
468       initCredentials(job);
469     }
470   }
471 
472   public static void initCredentials(Job job) throws IOException {
473     UserProvider userProvider = UserProvider.instantiate(job.getConfiguration());
474     if (userProvider.isHadoopSecurityEnabled()) {
475       // propagate delegation related props from launcher job to MR job
476       if (System.getenv("HADOOP_TOKEN_FILE_LOCATION") != null) {
477         job.getConfiguration().set("mapreduce.job.credentials.binary",
478                                    System.getenv("HADOOP_TOKEN_FILE_LOCATION"));
479       }
480     }
481 
482     if (userProvider.isHBaseSecurityEnabled()) {
483       try {
484         // init credentials for remote cluster
485         String quorumAddress = job.getConfiguration().get(TableOutputFormat.QUORUM_ADDRESS);
486         User user = userProvider.getCurrent();
487         if (quorumAddress != null) {
488           Configuration peerConf = HBaseConfiguration.create(job.getConfiguration());
489           ZKUtil.applyClusterKeyToConf(peerConf, quorumAddress);
490           Connection peerConn = ConnectionFactory.createConnection(peerConf);
491           try {
492             TokenUtil.addTokenForJob(peerConn, user, job);
493           } finally {
494             peerConn.close();
495           }
496         }
497 
498         Connection conn = ConnectionFactory.createConnection(job.getConfiguration());
499         try {
500           TokenUtil.addTokenForJob(conn, user, job);
501         } finally {
502           conn.close();
503         }
504       } catch (InterruptedException ie) {
505         LOG.info("Interrupted obtaining user authentication token");
506         Thread.currentThread().interrupt();
507       }
508     }
509   }
510 
511   /**
512    * Obtain an authentication token, for the specified cluster, on behalf of the current user
513    * and add it to the credentials for the given map reduce job.
514    *
515    * The quorumAddress is the key to the ZK ensemble, which contains:
516    * hbase.zookeeper.quorum, hbase.zookeeper.client.port and
517    * zookeeper.znode.parent
518    *
519    * @param job The job that requires the permission.
520    * @param quorumAddress string that contains the 3 required configuratins
521    * @throws IOException When the authentication token cannot be obtained.
522    */
523   public static void initCredentialsForCluster(Job job, String quorumAddress)
524       throws IOException {
525     UserProvider userProvider = UserProvider.instantiate(job.getConfiguration());
526     if (userProvider.isHBaseSecurityEnabled()) {
527       try {
528         Configuration peerConf = HBaseConfiguration.create(job.getConfiguration());
529         ZKUtil.applyClusterKeyToConf(peerConf, quorumAddress);
530         Connection peerConn = ConnectionFactory.createConnection(peerConf);
531         try {
532           TokenUtil.addTokenForJob(peerConn, userProvider.getCurrent(), job);
533         } finally {
534           peerConn.close();
535         }
536       } catch (InterruptedException e) {
537         LOG.info("Interrupted obtaining user authentication token");
538         Thread.interrupted();
539       }
540     }
541   }
542 
543   /**
544    * Writes the given scan into a Base64 encoded string.
545    *
546    * @param scan  The scan to write out.
547    * @return The scan saved in a Base64 encoded string.
548    * @throws IOException When writing the scan fails.
549    */
550   static String convertScanToString(Scan scan) throws IOException {
551     ClientProtos.Scan proto = ProtobufUtil.toScan(scan);
552     return Base64.encodeBytes(proto.toByteArray());
553   }
554 
555   /**
556    * Converts the given Base64 string back into a Scan instance.
557    *
558    * @param base64  The scan details.
559    * @return The newly created Scan instance.
560    * @throws IOException When reading the scan instance fails.
561    */
562   static Scan convertStringToScan(String base64) throws IOException {
563     byte [] decoded = Base64.decode(base64);
564     ClientProtos.Scan scan;
565     try {
566       scan = ClientProtos.Scan.parseFrom(decoded);
567     } catch (InvalidProtocolBufferException ipbe) {
568       throw new IOException(ipbe);
569     }
570 
571     return ProtobufUtil.toScan(scan);
572   }
573 
574   /**
575    * Use this before submitting a TableReduce job. It will
576    * appropriately set up the JobConf.
577    *
578    * @param table  The output table.
579    * @param reducer  The reducer class to use.
580    * @param job  The current job to adjust.
581    * @throws IOException When determining the region count fails.
582    */
583   public static void initTableReducerJob(String table,
584     Class<? extends TableReducer> reducer, Job job)
585   throws IOException {
586     initTableReducerJob(table, reducer, job, null);
587   }
588 
589   /**
590    * Use this before submitting a TableReduce job. It will
591    * appropriately set up the JobConf.
592    *
593    * @param table  The output table.
594    * @param reducer  The reducer class to use.
595    * @param job  The current job to adjust.
596    * @param partitioner  Partitioner to use. Pass <code>null</code> to use
597    * default partitioner.
598    * @throws IOException When determining the region count fails.
599    */
600   public static void initTableReducerJob(String table,
601     Class<? extends TableReducer> reducer, Job job,
602     Class partitioner) throws IOException {
603     initTableReducerJob(table, reducer, job, partitioner, null, null, null);
604   }
605 
606   /**
607    * Use this before submitting a TableReduce job. It will
608    * appropriately set up the JobConf.
609    *
610    * @param table  The output table.
611    * @param reducer  The reducer class to use.
612    * @param job  The current job to adjust.  Make sure the passed job is
613    * carrying all necessary HBase configuration.
614    * @param partitioner  Partitioner to use. Pass <code>null</code> to use
615    * default partitioner.
616    * @param quorumAddress Distant cluster to write to; default is null for
617    * output to the cluster that is designated in <code>hbase-site.xml</code>.
618    * Set this String to the zookeeper ensemble of an alternate remote cluster
619    * when you would have the reduce write a cluster that is other than the
620    * default; e.g. copying tables between clusters, the source would be
621    * designated by <code>hbase-site.xml</code> and this param would have the
622    * ensemble address of the remote cluster.  The format to pass is particular.
623    * Pass <code> &lt;hbase.zookeeper.quorum&gt;:&lt;
624    *             hbase.zookeeper.client.port&gt;:&lt;zookeeper.znode.parent&gt;
625    * </code> such as <code>server,server2,server3:2181:/hbase</code>.
626    * @param serverClass redefined hbase.regionserver.class
627    * @param serverImpl redefined hbase.regionserver.impl
628    * @throws IOException When determining the region count fails.
629    */
630   public static void initTableReducerJob(String table,
631     Class<? extends TableReducer> reducer, Job job,
632     Class partitioner, String quorumAddress, String serverClass,
633     String serverImpl) throws IOException {
634     initTableReducerJob(table, reducer, job, partitioner, quorumAddress,
635         serverClass, serverImpl, true);
636   }
637 
638   /**
639    * Use this before submitting a TableReduce job. It will
640    * appropriately set up the JobConf.
641    *
642    * @param table  The output table.
643    * @param reducer  The reducer class to use.
644    * @param job  The current job to adjust.  Make sure the passed job is
645    * carrying all necessary HBase configuration.
646    * @param partitioner  Partitioner to use. Pass <code>null</code> to use
647    * default partitioner.
648    * @param quorumAddress Distant cluster to write to; default is null for
649    * output to the cluster that is designated in <code>hbase-site.xml</code>.
650    * Set this String to the zookeeper ensemble of an alternate remote cluster
651    * when you would have the reduce write a cluster that is other than the
652    * default; e.g. copying tables between clusters, the source would be
653    * designated by <code>hbase-site.xml</code> and this param would have the
654    * ensemble address of the remote cluster.  The format to pass is particular.
655    * Pass <code> &lt;hbase.zookeeper.quorum&gt;:&lt;
656    *             hbase.zookeeper.client.port&gt;:&lt;zookeeper.znode.parent&gt;
657    * </code> such as <code>server,server2,server3:2181:/hbase</code>.
658    * @param serverClass redefined hbase.regionserver.class
659    * @param serverImpl redefined hbase.regionserver.impl
660    * @param addDependencyJars upload HBase jars and jars for any of the configured
661    *           job classes via the distributed cache (tmpjars).
662    * @throws IOException When determining the region count fails.
663    */
664   public static void initTableReducerJob(String table,
665     Class<? extends TableReducer> reducer, Job job,
666     Class partitioner, String quorumAddress, String serverClass,
667     String serverImpl, boolean addDependencyJars) throws IOException {
668 
669     Configuration conf = job.getConfiguration();
670     HBaseConfiguration.merge(conf, HBaseConfiguration.create(conf));
671     job.setOutputFormatClass(TableOutputFormat.class);
672     if (reducer != null) job.setReducerClass(reducer);
673     conf.set(TableOutputFormat.OUTPUT_TABLE, table);
674     conf.setStrings("io.serializations", conf.get("io.serializations"),
675         MutationSerialization.class.getName(), ResultSerialization.class.getName());
676     // If passed a quorum/ensemble address, pass it on to TableOutputFormat.
677     if (quorumAddress != null) {
678       // Calling this will validate the format
679       ZKUtil.transformClusterKey(quorumAddress);
680       conf.set(TableOutputFormat.QUORUM_ADDRESS,quorumAddress);
681     }
682     if (serverClass != null && serverImpl != null) {
683       conf.set(TableOutputFormat.REGION_SERVER_CLASS, serverClass);
684       conf.set(TableOutputFormat.REGION_SERVER_IMPL, serverImpl);
685     }
686     job.setOutputKeyClass(ImmutableBytesWritable.class);
687     job.setOutputValueClass(Writable.class);
688     if (partitioner == HRegionPartitioner.class) {
689       job.setPartitionerClass(HRegionPartitioner.class);
690       int regions = MetaTableAccessor.getRegionCount(conf, TableName.valueOf(table));
691       if (job.getNumReduceTasks() > regions) {
692         job.setNumReduceTasks(regions);
693       }
694     } else if (partitioner != null) {
695       job.setPartitionerClass(partitioner);
696     }
697 
698     if (addDependencyJars) {
699       addDependencyJars(job);
700     }
701 
702     initCredentials(job);
703   }
704 
705   /**
706    * Ensures that the given number of reduce tasks for the given job
707    * configuration does not exceed the number of regions for the given table.
708    *
709    * @param table  The table to get the region count for.
710    * @param job  The current job to adjust.
711    * @throws IOException When retrieving the table details fails.
712    */
713   public static void limitNumReduceTasks(String table, Job job)
714   throws IOException {
715     int regions =
716       MetaTableAccessor.getRegionCount(job.getConfiguration(), TableName.valueOf(table));
717     if (job.getNumReduceTasks() > regions)
718       job.setNumReduceTasks(regions);
719   }
720 
721   /**
722    * Sets the number of reduce tasks for the given job configuration to the
723    * number of regions the given table has.
724    *
725    * @param table  The table to get the region count for.
726    * @param job  The current job to adjust.
727    * @throws IOException When retrieving the table details fails.
728    */
729   public static void setNumReduceTasks(String table, Job job)
730   throws IOException {
731     job.setNumReduceTasks(MetaTableAccessor.getRegionCount(job.getConfiguration(),
732        TableName.valueOf(table)));
733   }
734 
735   /**
736    * Sets the number of rows to return and cache with each scanner iteration.
737    * Higher caching values will enable faster mapreduce jobs at the expense of
738    * requiring more heap to contain the cached rows.
739    *
740    * @param job The current job to adjust.
741    * @param batchSize The number of rows to return in batch with each scanner
742    * iteration.
743    */
744   public static void setScannerCaching(Job job, int batchSize) {
745     job.getConfiguration().setInt("hbase.client.scanner.caching", batchSize);
746   }
747 
748   /**
749    * Add HBase and its dependencies (only) to the job configuration.
750    * <p>
751    * This is intended as a low-level API, facilitating code reuse between this
752    * class and its mapred counterpart. It also of use to extenral tools that
753    * need to build a MapReduce job that interacts with HBase but want
754    * fine-grained control over the jars shipped to the cluster.
755    * </p>
756    * @param conf The Configuration object to extend with dependencies.
757    * @see org.apache.hadoop.hbase.mapred.TableMapReduceUtil
758    * @see <a href="https://issues.apache.org/jira/browse/PIG-3285">PIG-3285</a>
759    */
760   public static void addHBaseDependencyJars(Configuration conf) throws IOException {
761     addDependencyJars(conf,
762       // explicitly pull a class from each module
763       org.apache.hadoop.hbase.HConstants.class,                      // hbase-common
764       org.apache.hadoop.hbase.protobuf.generated.ClientProtos.class, // hbase-protocol
765       org.apache.hadoop.hbase.client.Put.class,                      // hbase-client
766       org.apache.hadoop.hbase.CompatibilityFactory.class,            // hbase-hadoop-compat
767       org.apache.hadoop.hbase.mapreduce.TableMapper.class,           // hbase-server
768       // pull necessary dependencies
769       org.apache.zookeeper.ZooKeeper.class,
770       io.netty.channel.Channel.class,
771       com.google.protobuf.Message.class,
772       com.google.common.collect.Lists.class,
773       org.apache.htrace.Trace.class,
774       com.yammer.metrics.core.MetricsRegistry.class);
775   }
776 
777   /**
778    * Returns a classpath string built from the content of the "tmpjars" value in {@code conf}.
779    * Also exposed to shell scripts via `bin/hbase mapredcp`.
780    */
781   public static String buildDependencyClasspath(Configuration conf) {
782     if (conf == null) {
783       throw new IllegalArgumentException("Must provide a configuration object.");
784     }
785     Set<String> paths = new HashSet<String>(conf.getStringCollection("tmpjars"));
786     if (paths.size() == 0) {
787       throw new IllegalArgumentException("Configuration contains no tmpjars.");
788     }
789     StringBuilder sb = new StringBuilder();
790     for (String s : paths) {
791       // entries can take the form 'file:/path/to/file.jar'.
792       int idx = s.indexOf(":");
793       if (idx != -1) s = s.substring(idx + 1);
794       if (sb.length() > 0) sb.append(File.pathSeparator);
795       sb.append(s);
796     }
797     return sb.toString();
798   }
799 
800   /**
801    * Add the HBase dependency jars as well as jars for any of the configured
802    * job classes to the job configuration, so that JobClient will ship them
803    * to the cluster and add them to the DistributedCache.
804    */
805   public static void addDependencyJars(Job job) throws IOException {
806     addHBaseDependencyJars(job.getConfiguration());
807     try {
808       addDependencyJars(job.getConfiguration(),
809           // when making changes here, consider also mapred.TableMapReduceUtil
810           // pull job classes
811           job.getMapOutputKeyClass(),
812           job.getMapOutputValueClass(),
813           job.getInputFormatClass(),
814           job.getOutputKeyClass(),
815           job.getOutputValueClass(),
816           job.getOutputFormatClass(),
817           job.getPartitionerClass(),
818           job.getCombinerClass());
819     } catch (ClassNotFoundException e) {
820       throw new IOException(e);
821     }
822   }
823 
824   /**
825    * Add the jars containing the given classes to the job's configuration
826    * such that JobClient will ship them to the cluster and add them to
827    * the DistributedCache.
828    */
829   public static void addDependencyJars(Configuration conf,
830       Class<?>... classes) throws IOException {
831 
832     FileSystem localFs = FileSystem.getLocal(conf);
833     Set<String> jars = new HashSet<String>();
834     // Add jars that are already in the tmpjars variable
835     jars.addAll(conf.getStringCollection("tmpjars"));
836 
837     // add jars as we find them to a map of contents jar name so that we can avoid
838     // creating new jars for classes that have already been packaged.
839     Map<String, String> packagedClasses = new HashMap<String, String>();
840 
841     // Add jars containing the specified classes
842     for (Class<?> clazz : classes) {
843       if (clazz == null) continue;
844 
845       Path path = findOrCreateJar(clazz, localFs, packagedClasses);
846       if (path == null) {
847         LOG.warn("Could not find jar for class " + clazz +
848                  " in order to ship it to the cluster.");
849         continue;
850       }
851       if (!localFs.exists(path)) {
852         LOG.warn("Could not validate jar file " + path + " for class "
853                  + clazz);
854         continue;
855       }
856       jars.add(path.toString());
857     }
858     if (jars.isEmpty()) return;
859 
860     conf.set("tmpjars", StringUtils.arrayToString(jars.toArray(new String[jars.size()])));
861   }
862 
863   /**
864    * Finds the Jar for a class or creates it if it doesn't exist. If the class is in
865    * a directory in the classpath, it creates a Jar on the fly with the
866    * contents of the directory and returns the path to that Jar. If a Jar is
867    * created, it is created in the system temporary directory. Otherwise,
868    * returns an existing jar that contains a class of the same name. Maintains
869    * a mapping from jar contents to the tmp jar created.
870    * @param my_class the class to find.
871    * @param fs the FileSystem with which to qualify the returned path.
872    * @param packagedClasses a map of class name to path.
873    * @return a jar file that contains the class.
874    * @throws IOException
875    */
876   private static Path findOrCreateJar(Class<?> my_class, FileSystem fs,
877       Map<String, String> packagedClasses)
878   throws IOException {
879     // attempt to locate an existing jar for the class.
880     String jar = findContainingJar(my_class, packagedClasses);
881     if (null == jar || jar.isEmpty()) {
882       jar = getJar(my_class);
883       updateMap(jar, packagedClasses);
884     }
885 
886     if (null == jar || jar.isEmpty()) {
887       return null;
888     }
889 
890     LOG.debug(String.format("For class %s, using jar %s", my_class.getName(), jar));
891     return new Path(jar).makeQualified(fs);
892   }
893 
894   /**
895    * Add entries to <code>packagedClasses</code> corresponding to class files
896    * contained in <code>jar</code>.
897    * @param jar The jar who's content to list.
898    * @param packagedClasses map[class -> jar]
899    */
900   private static void updateMap(String jar, Map<String, String> packagedClasses) throws IOException {
901     if (null == jar || jar.isEmpty()) {
902       return;
903     }
904     ZipFile zip = null;
905     try {
906       zip = new ZipFile(jar);
907       for (Enumeration<? extends ZipEntry> iter = zip.entries(); iter.hasMoreElements();) {
908         ZipEntry entry = iter.nextElement();
909         if (entry.getName().endsWith("class")) {
910           packagedClasses.put(entry.getName(), jar);
911         }
912       }
913     } finally {
914       if (null != zip) zip.close();
915     }
916   }
917 
918   /**
919    * Find a jar that contains a class of the same name, if any. It will return
920    * a jar file, even if that is not the first thing on the class path that
921    * has a class with the same name. Looks first on the classpath and then in
922    * the <code>packagedClasses</code> map.
923    * @param my_class the class to find.
924    * @return a jar file that contains the class, or null.
925    * @throws IOException
926    */
927   private static String findContainingJar(Class<?> my_class, Map<String, String> packagedClasses)
928       throws IOException {
929     ClassLoader loader = my_class.getClassLoader();
930 
931     String class_file = my_class.getName().replaceAll("\\.", "/") + ".class";
932 
933     if (loader != null) {
934       // first search the classpath
935       for (Enumeration<URL> itr = loader.getResources(class_file); itr.hasMoreElements();) {
936         URL url = itr.nextElement();
937         if ("jar".equals(url.getProtocol())) {
938           String toReturn = url.getPath();
939           if (toReturn.startsWith("file:")) {
940             toReturn = toReturn.substring("file:".length());
941           }
942           // URLDecoder is a misnamed class, since it actually decodes
943           // x-www-form-urlencoded MIME type rather than actual
944           // URL encoding (which the file path has). Therefore it would
945           // decode +s to ' 's which is incorrect (spaces are actually
946           // either unencoded or encoded as "%20"). Replace +s first, so
947           // that they are kept sacred during the decoding process.
948           toReturn = toReturn.replaceAll("\\+", "%2B");
949           toReturn = URLDecoder.decode(toReturn, "UTF-8");
950           return toReturn.replaceAll("!.*$", "");
951         }
952       }
953     }
954 
955     // now look in any jars we've packaged using JarFinder. Returns null when
956     // no jar is found.
957     return packagedClasses.get(class_file);
958   }
959 
960   /**
961    * Invoke 'getJar' on a custom JarFinder implementation. Useful for some job
962    * configuration contexts (HBASE-8140) and also for testing on MRv2.
963    * check if we have HADOOP-9426.
964    * @param my_class the class to find.
965    * @return a jar file that contains the class, or null.
966    */
967   private static String getJar(Class<?> my_class) {
968     String ret = null;
969     try {
970       ret = JarFinder.getJar(my_class);
971     } catch (Exception e) {
972       // toss all other exceptions, related to reflection failure
973       throw new RuntimeException("getJar invocation failed.", e);
974     }
975 
976     return ret;
977   }
978 }