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