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