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