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