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