View Javadoc

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