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