View Javadoc

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