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