| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package org.apache.hadoop.hbase.mapreduce; |
| |
| import com.codahale.metrics.MetricRegistry; |
| import java.io.File; |
| import java.io.IOException; |
| import java.net.URL; |
| import java.net.URLDecoder; |
| import java.util.ArrayList; |
| import java.util.Base64; |
| import java.util.Collection; |
| import java.util.Enumeration; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.zip.ZipEntry; |
| import java.util.zip.ZipFile; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.hbase.HBaseConfiguration; |
| import org.apache.hadoop.hbase.HConstants; |
| import org.apache.hadoop.hbase.TableName; |
| import org.apache.hadoop.hbase.client.Connection; |
| import org.apache.hadoop.hbase.client.ConnectionFactory; |
| import org.apache.hadoop.hbase.client.Put; |
| import org.apache.hadoop.hbase.client.RegionLocator; |
| import org.apache.hadoop.hbase.client.Scan; |
| import org.apache.hadoop.hbase.io.ImmutableBytesWritable; |
| import org.apache.hadoop.hbase.security.User; |
| import org.apache.hadoop.hbase.security.UserProvider; |
| import org.apache.hadoop.hbase.security.token.TokenUtil; |
| import org.apache.hadoop.hbase.util.Bytes; |
| import org.apache.hadoop.hbase.util.RegionSplitter; |
| import org.apache.hadoop.hbase.zookeeper.ZKConfig; |
| import org.apache.hadoop.io.Writable; |
| import org.apache.hadoop.mapreduce.InputFormat; |
| import org.apache.hadoop.mapreduce.Job; |
| import org.apache.hadoop.util.StringUtils; |
| import org.apache.yetus.audience.InterfaceAudience; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil; |
| import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos; |
| |
| /** |
| * Utility for {@link TableMapper} and {@link TableReducer} |
| */ |
| @SuppressWarnings({ "rawtypes", "unchecked" }) |
| @InterfaceAudience.Public |
| public class TableMapReduceUtil { |
| private static final Logger LOG = LoggerFactory.getLogger(TableMapReduceUtil.class); |
| public static final String TABLE_INPUT_CLASS_KEY = "hbase.table.input.class"; |
| |
| /** |
| * Use this before submitting a TableMap job. It will appropriately set up the job. |
| * @param table The table name to read from. |
| * @param scan The scan instance with the columns, time range etc. |
| * @param mapper The mapper class to use. |
| * @param outputKeyClass The class of the output key. |
| * @param outputValueClass The class of the output value. |
| * @param job The current job to adjust. Make sure the passed job is carrying all |
| * necessary HBase configuration. |
| * @throws IOException When setting up the details fails. |
| */ |
| public static void initTableMapperJob(String table, Scan scan, |
| Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, |
| Job job) throws IOException { |
| initTableMapperJob(table, scan, mapper, outputKeyClass, outputValueClass, job, true); |
| } |
| |
| /** |
| * Use this before submitting a TableMap job. It will appropriately set up the job. |
| * @param table The table name to read from. |
| * @param scan The scan instance with the columns, time range etc. |
| * @param mapper The mapper class to use. |
| * @param outputKeyClass The class of the output key. |
| * @param outputValueClass The class of the output value. |
| * @param job The current job to adjust. Make sure the passed job is carrying all |
| * necessary HBase configuration. |
| * @throws IOException When setting up the details fails. |
| */ |
| public static void initTableMapperJob(TableName table, Scan scan, |
| Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, |
| Job job) throws IOException { |
| initTableMapperJob(table.getNameAsString(), scan, mapper, outputKeyClass, outputValueClass, job, |
| true); |
| } |
| |
| /** |
| * Use this before submitting a TableMap job. It will appropriately set up the job. |
| * @param table Binary representation of the table name to read from. |
| * @param scan The scan instance with the columns, time range etc. |
| * @param mapper The mapper class to use. |
| * @param outputKeyClass The class of the output key. |
| * @param outputValueClass The class of the output value. |
| * @param job The current job to adjust. Make sure the passed job is carrying all |
| * necessary HBase configuration. |
| * @throws IOException When setting up the details fails. |
| */ |
| public static void initTableMapperJob(byte[] table, Scan scan, |
| Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, |
| Job job) throws IOException { |
| initTableMapperJob(Bytes.toString(table), scan, mapper, outputKeyClass, outputValueClass, job, |
| true); |
| } |
| |
| /** |
| * Use this before submitting a TableMap job. It will appropriately set up the job. |
| * @param table The table name to read from. |
| * @param scan The scan instance with the columns, time range etc. |
| * @param mapper The mapper class to use. |
| * @param outputKeyClass The class of the output key. |
| * @param outputValueClass The class of the output value. |
| * @param job The current job to adjust. Make sure the passed job is carrying all |
| * necessary HBase configuration. |
| * @param addDependencyJars upload HBase jars and jars for any of the configured job classes via |
| * the distributed cache (tmpjars). |
| * @throws IOException When setting up the details fails. |
| */ |
| public static void initTableMapperJob(String table, Scan scan, |
| Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, |
| Job job, boolean addDependencyJars, Class<? extends InputFormat> inputFormatClass) |
| throws IOException { |
| initTableMapperJob(table, scan, mapper, outputKeyClass, outputValueClass, job, |
| addDependencyJars, true, inputFormatClass); |
| } |
| |
| /** |
| * Use this before submitting a TableMap job. It will appropriately set up the job. |
| * @param table The table name to read from. |
| * @param scan The scan instance with the columns, time range etc. |
| * @param mapper The mapper class to use. |
| * @param outputKeyClass The class of the output key. |
| * @param outputValueClass The class of the output value. |
| * @param job The current job to adjust. Make sure the passed job is carrying all |
| * necessary HBase configuration. |
| * @param addDependencyJars upload HBase jars and jars for any of the configured job classes via |
| * the distributed cache (tmpjars). |
| * @param initCredentials whether to initialize hbase auth credentials for the job |
| * @param inputFormatClass the input format |
| * @throws IOException When setting up the details fails. |
| */ |
| public static void initTableMapperJob(String table, Scan scan, |
| Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, |
| Job job, boolean addDependencyJars, boolean initCredentials, |
| Class<? extends InputFormat> inputFormatClass) throws IOException { |
| job.setInputFormatClass(inputFormatClass); |
| if (outputValueClass != null) job.setMapOutputValueClass(outputValueClass); |
| if (outputKeyClass != null) job.setMapOutputKeyClass(outputKeyClass); |
| job.setMapperClass(mapper); |
| if (Put.class.equals(outputValueClass)) { |
| job.setCombinerClass(PutCombiner.class); |
| } |
| Configuration conf = job.getConfiguration(); |
| HBaseConfiguration.merge(conf, HBaseConfiguration.create(conf)); |
| conf.set(TableInputFormat.INPUT_TABLE, table); |
| conf.set(TableInputFormat.SCAN, convertScanToString(scan)); |
| conf.setStrings("io.serializations", conf.get("io.serializations"), |
| MutationSerialization.class.getName(), ResultSerialization.class.getName(), |
| CellSerialization.class.getName()); |
| if (addDependencyJars) { |
| addDependencyJars(job); |
| } |
| if (initCredentials) { |
| initCredentials(job); |
| } |
| } |
| |
| /** |
| * Use this before submitting a TableMap job. It will appropriately set up the job. |
| * @param table Binary representation of the table name to read from. |
| * @param scan The scan instance with the columns, time range etc. |
| * @param mapper The mapper class to use. |
| * @param outputKeyClass The class of the output key. |
| * @param outputValueClass The class of the output value. |
| * @param job The current job to adjust. Make sure the passed job is carrying all |
| * necessary HBase configuration. |
| * @param addDependencyJars upload HBase jars and jars for any of the configured job classes via |
| * the distributed cache (tmpjars). |
| * @param inputFormatClass The class of the input format |
| * @throws IOException When setting up the details fails. |
| */ |
| public static void initTableMapperJob(byte[] table, Scan scan, |
| Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, |
| Job job, boolean addDependencyJars, Class<? extends InputFormat> inputFormatClass) |
| throws IOException { |
| initTableMapperJob(Bytes.toString(table), scan, mapper, outputKeyClass, outputValueClass, job, |
| addDependencyJars, inputFormatClass); |
| } |
| |
| /** |
| * Use this before submitting a TableMap job. It will appropriately set up the job. |
| * @param table Binary representation of the table name to read from. |
| * @param scan The scan instance with the columns, time range etc. |
| * @param mapper The mapper class to use. |
| * @param outputKeyClass The class of the output key. |
| * @param outputValueClass The class of the output value. |
| * @param job The current job to adjust. Make sure the passed job is carrying all |
| * necessary HBase configuration. |
| * @param addDependencyJars upload HBase jars and jars for any of the configured job classes via |
| * the distributed cache (tmpjars). |
| * @throws IOException When setting up the details fails. |
| */ |
| public static void initTableMapperJob(byte[] table, Scan scan, |
| Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, |
| Job job, boolean addDependencyJars) throws IOException { |
| initTableMapperJob(Bytes.toString(table), scan, mapper, outputKeyClass, outputValueClass, job, |
| addDependencyJars, getConfiguredInputFormat(job)); |
| } |
| |
| /** |
| * @return {@link TableInputFormat} .class unless Configuration has something else at |
| * {@link #TABLE_INPUT_CLASS_KEY}. |
| */ |
| private static Class<? extends InputFormat> getConfiguredInputFormat(Job job) { |
| return (Class<? extends InputFormat>) job.getConfiguration().getClass(TABLE_INPUT_CLASS_KEY, |
| TableInputFormat.class); |
| } |
| |
| /** |
| * Use this before submitting a TableMap job. It will appropriately set up the job. |
| * @param table The table name to read from. |
| * @param scan The scan instance with the columns, time range etc. |
| * @param mapper The mapper class to use. |
| * @param outputKeyClass The class of the output key. |
| * @param outputValueClass The class of the output value. |
| * @param job The current job to adjust. Make sure the passed job is carrying all |
| * necessary HBase configuration. |
| * @param addDependencyJars upload HBase jars and jars for any of the configured job classes via |
| * the distributed cache (tmpjars). |
| * @throws IOException When setting up the details fails. |
| */ |
| public static void initTableMapperJob(String table, Scan scan, |
| Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, |
| Job job, boolean addDependencyJars) throws IOException { |
| initTableMapperJob(table, scan, mapper, outputKeyClass, outputValueClass, job, |
| addDependencyJars, getConfiguredInputFormat(job)); |
| } |
| |
| /** |
| * Enable a basic on-heap cache for these jobs. Any BlockCache implementation based on direct |
| * memory will likely cause the map tasks to OOM when opening the region. This is done here |
| * instead of in TableSnapshotRegionRecordReader in case an advanced user wants to override this |
| * behavior in their job. |
| */ |
| public static void resetCacheConfig(Configuration conf) { |
| conf.setFloat(HConstants.HFILE_BLOCK_CACHE_SIZE_KEY, HConstants.HFILE_BLOCK_CACHE_SIZE_DEFAULT); |
| conf.setFloat(HConstants.BUCKET_CACHE_SIZE_KEY, 0f); |
| conf.unset(HConstants.BUCKET_CACHE_IOENGINE_KEY); |
| } |
| |
| /** |
| * Sets up the job for reading from one or more table snapshots, with one or more scans per |
| * snapshot. It bypasses hbase servers and read directly from snapshot files. |
| * @param snapshotScans map of snapshot name to scans on that snapshot. |
| * @param mapper The mapper class to use. |
| * @param outputKeyClass The class of the output key. |
| * @param outputValueClass The class of the output value. |
| * @param job The current job to adjust. Make sure the passed job is carrying all |
| * necessary HBase configuration. |
| * @param addDependencyJars upload HBase jars and jars for any of the configured job classes via |
| * the distributed cache (tmpjars). |
| */ |
| public static void initMultiTableSnapshotMapperJob(Map<String, Collection<Scan>> snapshotScans, |
| Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, |
| Job job, boolean addDependencyJars, Path tmpRestoreDir) throws IOException { |
| MultiTableSnapshotInputFormat.setInput(job.getConfiguration(), snapshotScans, tmpRestoreDir); |
| |
| job.setInputFormatClass(MultiTableSnapshotInputFormat.class); |
| if (outputValueClass != null) { |
| job.setMapOutputValueClass(outputValueClass); |
| } |
| if (outputKeyClass != null) { |
| job.setMapOutputKeyClass(outputKeyClass); |
| } |
| job.setMapperClass(mapper); |
| Configuration conf = job.getConfiguration(); |
| HBaseConfiguration.merge(conf, HBaseConfiguration.create(conf)); |
| |
| if (addDependencyJars) { |
| addDependencyJars(job); |
| addDependencyJarsForClasses(job.getConfiguration(), MetricRegistry.class); |
| } |
| |
| resetCacheConfig(job.getConfiguration()); |
| } |
| |
| /** |
| * Sets up the job for reading from a table snapshot. It bypasses hbase servers and read directly |
| * from snapshot files. |
| * @param snapshotName The name of the snapshot (of a table) to read from. |
| * @param scan The scan instance with the columns, time range etc. |
| * @param mapper The mapper class to use. |
| * @param outputKeyClass The class of the output key. |
| * @param outputValueClass The class of the output value. |
| * @param job The current job to adjust. Make sure the passed job is carrying all |
| * necessary HBase configuration. |
| * @param addDependencyJars upload HBase jars and jars for any of the configured job classes via |
| * the distributed cache (tmpjars). |
| * @param tmpRestoreDir a temporary directory to copy the snapshot files into. Current user |
| * should have write permissions to this directory, and this should not |
| * be a subdirectory of rootdir. After the job is finished, restore |
| * directory can be deleted. |
| * @throws IOException When setting up the details fails. |
| * @see TableSnapshotInputFormat |
| */ |
| public static void initTableSnapshotMapperJob(String snapshotName, Scan scan, |
| Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, |
| Job job, boolean addDependencyJars, Path tmpRestoreDir) throws IOException { |
| TableSnapshotInputFormat.setInput(job, snapshotName, tmpRestoreDir); |
| initTableMapperJob(snapshotName, scan, mapper, outputKeyClass, outputValueClass, job, |
| addDependencyJars, false, TableSnapshotInputFormat.class); |
| resetCacheConfig(job.getConfiguration()); |
| } |
| |
| /** |
| * Sets up the job for reading from a table snapshot. It bypasses hbase servers and read directly |
| * from snapshot files. |
| * @param snapshotName The name of the snapshot (of a table) to read from. |
| * @param scan The scan instance with the columns, time range etc. |
| * @param mapper The mapper class to use. |
| * @param outputKeyClass The class of the output key. |
| * @param outputValueClass The class of the output value. |
| * @param job The current job to adjust. Make sure the passed job is carrying all |
| * necessary HBase configuration. |
| * @param addDependencyJars upload HBase jars and jars for any of the configured job classes via |
| * the distributed cache (tmpjars). |
| * @param tmpRestoreDir a temporary directory to copy the snapshot files into. Current user |
| * should have write permissions to this directory, and this should not |
| * be a subdirectory of rootdir. After the job is finished, restore |
| * directory can be deleted. |
| * @param splitAlgo algorithm to split |
| * @param numSplitsPerRegion how many input splits to generate per one region |
| * @throws IOException When setting up the details fails. |
| * @see TableSnapshotInputFormat |
| */ |
| public static void initTableSnapshotMapperJob(String snapshotName, Scan scan, |
| Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, |
| Job job, boolean addDependencyJars, Path tmpRestoreDir, RegionSplitter.SplitAlgorithm splitAlgo, |
| int numSplitsPerRegion) throws IOException { |
| TableSnapshotInputFormat.setInput(job, snapshotName, tmpRestoreDir, splitAlgo, |
| numSplitsPerRegion); |
| initTableMapperJob(snapshotName, scan, mapper, outputKeyClass, outputValueClass, job, |
| addDependencyJars, false, TableSnapshotInputFormat.class); |
| resetCacheConfig(job.getConfiguration()); |
| } |
| |
| /** |
| * Use this before submitting a Multi TableMap job. It will appropriately set up the job. |
| * @param scans The list of {@link Scan} objects to read from. |
| * @param mapper The mapper class to use. |
| * @param outputKeyClass The class of the output key. |
| * @param outputValueClass The class of the output value. |
| * @param job The current job to adjust. Make sure the passed job is carrying all |
| * necessary HBase configuration. |
| * @throws IOException When setting up the details fails. |
| */ |
| public static void initTableMapperJob(List<Scan> scans, Class<? extends TableMapper> mapper, |
| Class<?> outputKeyClass, Class<?> outputValueClass, Job job) throws IOException { |
| initTableMapperJob(scans, mapper, outputKeyClass, outputValueClass, job, true); |
| } |
| |
| /** |
| * Use this before submitting a Multi TableMap job. It will appropriately set up the job. |
| * @param scans The list of {@link Scan} objects to read from. |
| * @param mapper The mapper class to use. |
| * @param outputKeyClass The class of the output key. |
| * @param outputValueClass The class of the output value. |
| * @param job The current job to adjust. Make sure the passed job is carrying all |
| * necessary HBase configuration. |
| * @param addDependencyJars upload HBase jars and jars for any of the configured job classes via |
| * the distributed cache (tmpjars). |
| * @throws IOException When setting up the details fails. |
| */ |
| public static void initTableMapperJob(List<Scan> scans, Class<? extends TableMapper> mapper, |
| Class<?> outputKeyClass, Class<?> outputValueClass, Job job, boolean addDependencyJars) |
| throws IOException { |
| initTableMapperJob(scans, mapper, outputKeyClass, outputValueClass, job, addDependencyJars, |
| true); |
| } |
| |
| /** |
| * Use this before submitting a Multi TableMap job. It will appropriately set up the job. |
| * @param scans The list of {@link Scan} objects to read from. |
| * @param mapper The mapper class to use. |
| * @param outputKeyClass The class of the output key. |
| * @param outputValueClass The class of the output value. |
| * @param job The current job to adjust. Make sure the passed job is carrying all |
| * necessary HBase configuration. |
| * @param addDependencyJars upload HBase jars and jars for any of the configured job classes via |
| * the distributed cache (tmpjars). |
| * @param initCredentials whether to initialize hbase auth credentials for the job |
| * @throws IOException When setting up the details fails. |
| */ |
| public static void initTableMapperJob(List<Scan> scans, Class<? extends TableMapper> mapper, |
| Class<?> outputKeyClass, Class<?> outputValueClass, Job job, boolean addDependencyJars, |
| boolean initCredentials) throws IOException { |
| job.setInputFormatClass(MultiTableInputFormat.class); |
| if (outputValueClass != null) { |
| job.setMapOutputValueClass(outputValueClass); |
| } |
| if (outputKeyClass != null) { |
| job.setMapOutputKeyClass(outputKeyClass); |
| } |
| job.setMapperClass(mapper); |
| Configuration conf = job.getConfiguration(); |
| HBaseConfiguration.merge(conf, HBaseConfiguration.create(conf)); |
| List<String> scanStrings = new ArrayList<>(); |
| |
| for (Scan scan : scans) { |
| scanStrings.add(convertScanToString(scan)); |
| } |
| job.getConfiguration().setStrings(MultiTableInputFormat.SCANS, |
| scanStrings.toArray(new String[scanStrings.size()])); |
| |
| if (addDependencyJars) { |
| addDependencyJars(job); |
| } |
| |
| if (initCredentials) { |
| initCredentials(job); |
| } |
| } |
| |
| public static void initCredentials(Job job) throws IOException { |
| UserProvider userProvider = UserProvider.instantiate(job.getConfiguration()); |
| if (userProvider.isHadoopSecurityEnabled()) { |
| // propagate delegation related props from launcher job to MR job |
| if (System.getenv("HADOOP_TOKEN_FILE_LOCATION") != null) { |
| job.getConfiguration().set("mapreduce.job.credentials.binary", |
| System.getenv("HADOOP_TOKEN_FILE_LOCATION")); |
| } |
| } |
| |
| if (userProvider.isHBaseSecurityEnabled()) { |
| try { |
| // init credentials for remote cluster |
| String quorumAddress = job.getConfiguration().get(TableOutputFormat.QUORUM_ADDRESS); |
| User user = userProvider.getCurrent(); |
| if (quorumAddress != null) { |
| Configuration peerConf = HBaseConfiguration.createClusterConf(job.getConfiguration(), |
| quorumAddress, TableOutputFormat.OUTPUT_CONF_PREFIX); |
| Connection peerConn = ConnectionFactory.createConnection(peerConf); |
| try { |
| TokenUtil.addTokenForJob(peerConn, user, job); |
| } finally { |
| peerConn.close(); |
| } |
| } |
| |
| Connection conn = ConnectionFactory.createConnection(job.getConfiguration()); |
| try { |
| TokenUtil.addTokenForJob(conn, user, job); |
| } finally { |
| conn.close(); |
| } |
| } catch (InterruptedException ie) { |
| LOG.info("Interrupted obtaining user authentication token"); |
| Thread.currentThread().interrupt(); |
| } |
| } |
| } |
| |
| /** |
| * Obtain an authentication token, for the specified cluster, on behalf of the current user and |
| * add it to the credentials for the given map reduce job. |
| * @param job The job that requires the permission. |
| * @param conf The configuration to use in connecting to the peer cluster |
| * @throws IOException When the authentication token cannot be obtained. |
| */ |
| public static void initCredentialsForCluster(Job job, Configuration conf) throws IOException { |
| UserProvider userProvider = UserProvider.instantiate(conf); |
| if (userProvider.isHBaseSecurityEnabled()) { |
| try { |
| Connection peerConn = ConnectionFactory.createConnection(conf); |
| try { |
| TokenUtil.addTokenForJob(peerConn, userProvider.getCurrent(), job); |
| } finally { |
| peerConn.close(); |
| } |
| } catch (InterruptedException e) { |
| LOG.info("Interrupted obtaining user authentication token"); |
| Thread.interrupted(); |
| } |
| } |
| } |
| |
| /** |
| * Writes the given scan into a Base64 encoded string. |
| * @param scan The scan to write out. |
| * @return The scan saved in a Base64 encoded string. |
| * @throws IOException When writing the scan fails. |
| */ |
| public static String convertScanToString(Scan scan) throws IOException { |
| ClientProtos.Scan proto = ProtobufUtil.toScan(scan); |
| return Bytes.toString(Base64.getEncoder().encode(proto.toByteArray())); |
| } |
| |
| /** |
| * Converts the given Base64 string back into a Scan instance. |
| * @param base64 The scan details. |
| * @return The newly created Scan instance. |
| * @throws IOException When reading the scan instance fails. |
| */ |
| public static Scan convertStringToScan(String base64) throws IOException { |
| byte[] decoded = Base64.getDecoder().decode(base64); |
| return ProtobufUtil.toScan(ClientProtos.Scan.parseFrom(decoded)); |
| } |
| |
| /** |
| * Use this before submitting a TableReduce job. It will appropriately set up the JobConf. |
| * @param table The output table. |
| * @param reducer The reducer class to use. |
| * @param job The current job to adjust. |
| * @throws IOException When determining the region count fails. |
| */ |
| public static void initTableReducerJob(String table, Class<? extends TableReducer> reducer, |
| Job job) throws IOException { |
| initTableReducerJob(table, reducer, job, null); |
| } |
| |
| /** |
| * Use this before submitting a TableReduce job. It will appropriately set up the JobConf. |
| * @param table The output table. |
| * @param reducer The reducer class to use. |
| * @param job The current job to adjust. |
| * @param partitioner Partitioner to use. Pass <code>null</code> to use default partitioner. |
| * @throws IOException When determining the region count fails. |
| */ |
| public static void initTableReducerJob(String table, Class<? extends TableReducer> reducer, |
| Job job, Class partitioner) throws IOException { |
| initTableReducerJob(table, reducer, job, partitioner, null, null, null); |
| } |
| |
| /** |
| * Use this before submitting a TableReduce job. It will appropriately set up the JobConf. |
| * @param table The output table. |
| * @param reducer The reducer class to use. |
| * @param job The current job to adjust. Make sure the passed job is carrying all |
| * necessary HBase configuration. |
| * @param partitioner Partitioner to use. Pass <code>null</code> to use default partitioner. |
| * @param quorumAddress Distant cluster to write to; default is null for output to the cluster |
| * that is designated in <code>hbase-site.xml</code>. Set this String to the |
| * zookeeper ensemble of an alternate remote cluster when you would have the |
| * reduce write a cluster that is other than the default; e.g. copying tables |
| * between clusters, the source would be designated by |
| * <code>hbase-site.xml</code> and this param would have the ensemble address |
| * of the remote cluster. The format to pass is particular. Pass |
| * <code> <hbase.zookeeper.quorum>:< |
| * hbase.zookeeper.client.port>:<zookeeper.znode.parent> |
| * </code> such as <code>server,server2,server3:2181:/hbase</code>. |
| * @param serverClass redefined hbase.regionserver.class |
| * @param serverImpl redefined hbase.regionserver.impl |
| * @throws IOException When determining the region count fails. |
| */ |
| public static void initTableReducerJob(String table, Class<? extends TableReducer> reducer, |
| Job job, Class partitioner, String quorumAddress, String serverClass, String serverImpl) |
| throws IOException { |
| initTableReducerJob(table, reducer, job, partitioner, quorumAddress, serverClass, serverImpl, |
| true); |
| } |
| |
| /** |
| * Use this before submitting a TableReduce job. It will appropriately set up the JobConf. |
| * @param table The output table. |
| * @param reducer The reducer class to use. |
| * @param job The current job to adjust. Make sure the passed job is carrying all |
| * necessary HBase configuration. |
| * @param partitioner Partitioner to use. Pass <code>null</code> to use default partitioner. |
| * @param quorumAddress Distant cluster to write to; default is null for output to the cluster |
| * that is designated in <code>hbase-site.xml</code>. Set this String to |
| * the zookeeper ensemble of an alternate remote cluster when you would |
| * have the reduce write a cluster that is other than the default; e.g. |
| * copying tables between clusters, the source would be designated by |
| * <code>hbase-site.xml</code> and this param would have the ensemble |
| * address of the remote cluster. The format to pass is particular. Pass |
| * <code> <hbase.zookeeper.quorum>:< |
| * hbase.zookeeper.client.port>:<zookeeper.znode.parent> |
| * </code> such as <code>server,server2,server3:2181:/hbase</code>. |
| * @param serverClass redefined hbase.regionserver.class |
| * @param serverImpl redefined hbase.regionserver.impl |
| * @param addDependencyJars upload HBase jars and jars for any of the configured job classes via |
| * the distributed cache (tmpjars). |
| * @throws IOException When determining the region count fails. |
| */ |
| public static void initTableReducerJob(String table, Class<? extends TableReducer> reducer, |
| Job job, Class partitioner, String quorumAddress, String serverClass, String serverImpl, |
| boolean addDependencyJars) throws IOException { |
| |
| Configuration conf = job.getConfiguration(); |
| HBaseConfiguration.merge(conf, HBaseConfiguration.create(conf)); |
| job.setOutputFormatClass(TableOutputFormat.class); |
| if (reducer != null) job.setReducerClass(reducer); |
| conf.set(TableOutputFormat.OUTPUT_TABLE, table); |
| conf.setStrings("io.serializations", conf.get("io.serializations"), |
| MutationSerialization.class.getName(), ResultSerialization.class.getName()); |
| // If passed a quorum/ensemble address, pass it on to TableOutputFormat. |
| if (quorumAddress != null) { |
| // Calling this will validate the format |
| ZKConfig.validateClusterKey(quorumAddress); |
| conf.set(TableOutputFormat.QUORUM_ADDRESS, quorumAddress); |
| } |
| if (serverClass != null && serverImpl != null) { |
| conf.set(TableOutputFormat.REGION_SERVER_CLASS, serverClass); |
| conf.set(TableOutputFormat.REGION_SERVER_IMPL, serverImpl); |
| } |
| job.setOutputKeyClass(ImmutableBytesWritable.class); |
| job.setOutputValueClass(Writable.class); |
| if (partitioner == HRegionPartitioner.class) { |
| job.setPartitionerClass(HRegionPartitioner.class); |
| int regions = getRegionCount(conf, TableName.valueOf(table)); |
| if (job.getNumReduceTasks() > regions) { |
| job.setNumReduceTasks(regions); |
| } |
| } else if (partitioner != null) { |
| job.setPartitionerClass(partitioner); |
| } |
| |
| if (addDependencyJars) { |
| addDependencyJars(job); |
| } |
| |
| initCredentials(job); |
| } |
| |
| /** |
| * Ensures that the given number of reduce tasks for the given job configuration does not exceed |
| * the number of regions for the given table. |
| * @param table The table to get the region count for. |
| * @param job The current job to adjust. |
| * @throws IOException When retrieving the table details fails. |
| */ |
| public static void limitNumReduceTasks(String table, Job job) throws IOException { |
| int regions = getRegionCount(job.getConfiguration(), TableName.valueOf(table)); |
| if (job.getNumReduceTasks() > regions) { |
| job.setNumReduceTasks(regions); |
| } |
| } |
| |
| /** |
| * Sets the number of reduce tasks for the given job configuration to the number of regions the |
| * given table has. |
| * @param table The table to get the region count for. |
| * @param job The current job to adjust. |
| * @throws IOException When retrieving the table details fails. |
| */ |
| public static void setNumReduceTasks(String table, Job job) throws IOException { |
| job.setNumReduceTasks(getRegionCount(job.getConfiguration(), TableName.valueOf(table))); |
| } |
| |
| /** |
| * Sets the number of rows to return and cache with each scanner iteration. Higher caching values |
| * will enable faster mapreduce jobs at the expense of requiring more heap to contain the cached |
| * rows. |
| * @param job The current job to adjust. |
| * @param batchSize The number of rows to return in batch with each scanner iteration. |
| */ |
| public static void setScannerCaching(Job job, int batchSize) { |
| job.getConfiguration().setInt("hbase.client.scanner.caching", batchSize); |
| } |
| |
| /** |
| * Add HBase and its dependencies (only) to the job configuration. |
| * <p> |
| * This is intended as a low-level API, facilitating code reuse between this class and its mapred |
| * counterpart. It also of use to external tools that need to build a MapReduce job that interacts |
| * with HBase but want fine-grained control over the jars shipped to the cluster. |
| * </p> |
| * @param conf The Configuration object to extend with dependencies. |
| * @see org.apache.hadoop.hbase.mapred.TableMapReduceUtil |
| * @see <a href="https://issues.apache.org/jira/browse/PIG-3285">PIG-3285</a> |
| */ |
| public static void addHBaseDependencyJars(Configuration conf) throws IOException { |
| addDependencyJarsForClasses(conf, |
| // explicitly pull a class from each module |
| org.apache.hadoop.hbase.HConstants.class, // hbase-common |
| org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.class, // hbase-protocol-shaded |
| org.apache.hadoop.hbase.client.Put.class, // hbase-client |
| org.apache.hadoop.hbase.ipc.RpcServer.class, // hbase-server |
| org.apache.hadoop.hbase.CompatibilityFactory.class, // hbase-hadoop-compat |
| org.apache.hadoop.hbase.mapreduce.JobUtil.class, // hbase-hadoop2-compat |
| org.apache.hadoop.hbase.mapreduce.TableMapper.class, // hbase-mapreduce |
| org.apache.hadoop.hbase.metrics.impl.FastLongHistogram.class, // hbase-metrics |
| org.apache.hadoop.hbase.metrics.Snapshot.class, // hbase-metrics-api |
| org.apache.hadoop.hbase.replication.ReplicationUtils.class, // hbase-replication |
| org.apache.hadoop.hbase.http.HttpServer.class, // hbase-http |
| org.apache.hadoop.hbase.procedure2.Procedure.class, // hbase-procedure |
| org.apache.hadoop.hbase.zookeeper.ZKWatcher.class, // hbase-zookeeper |
| org.apache.hbase.thirdparty.com.google.common.collect.Lists.class, // hb-shaded-miscellaneous |
| org.apache.hbase.thirdparty.com.google.gson.GsonBuilder.class, // hbase-shaded-gson |
| org.apache.hbase.thirdparty.com.google.protobuf.UnsafeByteOperations.class, // hb-sh-protobuf |
| org.apache.hbase.thirdparty.io.netty.channel.Channel.class, // hbase-shaded-netty |
| org.apache.hadoop.hbase.unsafe.HBasePlatformDependent.class, // hbase-unsafe |
| org.apache.zookeeper.ZooKeeper.class, // zookeeper |
| com.codahale.metrics.MetricRegistry.class, // metrics-core |
| org.apache.commons.lang3.ArrayUtils.class, // commons-lang |
| io.opentelemetry.api.trace.Span.class, // opentelemetry-api |
| io.opentelemetry.semconv.trace.attributes.SemanticAttributes.class, // opentelemetry-semconv |
| io.opentelemetry.context.Context.class); // opentelemetry-context |
| } |
| |
| /** |
| * Returns a classpath string built from the content of the "tmpjars" value in {@code conf}. Also |
| * exposed to shell scripts via `bin/hbase mapredcp`. |
| */ |
| public static String buildDependencyClasspath(Configuration conf) { |
| if (conf == null) { |
| throw new IllegalArgumentException("Must provide a configuration object."); |
| } |
| Set<String> paths = new HashSet<>(conf.getStringCollection("tmpjars")); |
| if (paths.isEmpty()) { |
| throw new IllegalArgumentException("Configuration contains no tmpjars."); |
| } |
| StringBuilder sb = new StringBuilder(); |
| for (String s : paths) { |
| // entries can take the form 'file:/path/to/file.jar'. |
| int idx = s.indexOf(":"); |
| if (idx != -1) s = s.substring(idx + 1); |
| if (sb.length() > 0) sb.append(File.pathSeparator); |
| sb.append(s); |
| } |
| return sb.toString(); |
| } |
| |
| /** |
| * Add the HBase dependency jars as well as jars for any of the configured job classes to the job |
| * configuration, so that JobClient will ship them to the cluster and add them to the |
| * DistributedCache. |
| */ |
| public static void addDependencyJars(Job job) throws IOException { |
| addHBaseDependencyJars(job.getConfiguration()); |
| try { |
| addDependencyJarsForClasses(job.getConfiguration(), |
| // when making changes here, consider also mapred.TableMapReduceUtil |
| // pull job classes |
| job.getMapOutputKeyClass(), job.getMapOutputValueClass(), job.getInputFormatClass(), |
| job.getOutputKeyClass(), job.getOutputValueClass(), job.getOutputFormatClass(), |
| job.getPartitionerClass(), job.getCombinerClass()); |
| } catch (ClassNotFoundException e) { |
| throw new IOException(e); |
| } |
| } |
| |
| /** |
| * Add the jars containing the given classes to the job's configuration such that JobClient will |
| * ship them to the cluster and add them to the DistributedCache. N.B. that this method at most |
| * adds one jar per class given. If there is more than one jar available containing a class with |
| * the same name as a given class, we don't define which of those jars might be chosen. |
| * @param conf The Hadoop Configuration to modify |
| * @param classes will add just those dependencies needed to find the given classes |
| * @throws IOException if an underlying library call fails. |
| */ |
| @InterfaceAudience.Private |
| public static void addDependencyJarsForClasses(Configuration conf, Class<?>... classes) |
| throws IOException { |
| |
| FileSystem localFs = FileSystem.getLocal(conf); |
| Set<String> jars = new HashSet<>(); |
| // Add jars that are already in the tmpjars variable |
| jars.addAll(conf.getStringCollection("tmpjars")); |
| |
| // add jars as we find them to a map of contents jar name so that we can avoid |
| // creating new jars for classes that have already been packaged. |
| Map<String, String> packagedClasses = new HashMap<>(); |
| |
| // Add jars containing the specified classes |
| for (Class<?> clazz : classes) { |
| if (clazz == null) continue; |
| |
| Path path = findOrCreateJar(clazz, localFs, packagedClasses); |
| if (path == null) { |
| LOG.warn("Could not find jar for class " + clazz + " in order to ship it to the cluster."); |
| continue; |
| } |
| if (!localFs.exists(path)) { |
| LOG.warn("Could not validate jar file " + path + " for class " + clazz); |
| continue; |
| } |
| jars.add(path.toString()); |
| } |
| if (jars.isEmpty()) return; |
| |
| conf.set("tmpjars", StringUtils.arrayToString(jars.toArray(new String[jars.size()]))); |
| } |
| |
| /** |
| * Finds the Jar for a class or creates it if it doesn't exist. If the class is in a directory in |
| * the classpath, it creates a Jar on the fly with the contents of the directory and returns the |
| * path to that Jar. If a Jar is created, it is created in the system temporary directory. |
| * Otherwise, returns an existing jar that contains a class of the same name. Maintains a mapping |
| * from jar contents to the tmp jar created. |
| * @param my_class the class to find. |
| * @param fs the FileSystem with which to qualify the returned path. |
| * @param packagedClasses a map of class name to path. |
| * @return a jar file that contains the class. |
| */ |
| private static Path findOrCreateJar(Class<?> my_class, FileSystem fs, |
| Map<String, String> packagedClasses) throws IOException { |
| // attempt to locate an existing jar for the class. |
| String jar = findContainingJar(my_class, packagedClasses); |
| if (null == jar || jar.isEmpty()) { |
| jar = getJar(my_class); |
| updateMap(jar, packagedClasses); |
| } |
| |
| if (null == jar || jar.isEmpty()) { |
| return null; |
| } |
| |
| LOG.debug(String.format("For class %s, using jar %s", my_class.getName(), jar)); |
| return new Path(jar).makeQualified(fs.getUri(), fs.getWorkingDirectory()); |
| } |
| |
| /** |
| * Add entries to <code>packagedClasses</code> corresponding to class files contained in |
| * <code>jar</code>. |
| * @param jar The jar who's content to list. |
| * @param packagedClasses map[class -> jar] |
| */ |
| private static void updateMap(String jar, Map<String, String> packagedClasses) |
| throws IOException { |
| if (null == jar || jar.isEmpty()) { |
| return; |
| } |
| ZipFile zip = null; |
| try { |
| zip = new ZipFile(jar); |
| for (Enumeration<? extends ZipEntry> iter = zip.entries(); iter.hasMoreElements();) { |
| ZipEntry entry = iter.nextElement(); |
| if (entry.getName().endsWith("class")) { |
| packagedClasses.put(entry.getName(), jar); |
| } |
| } |
| } finally { |
| if (null != zip) zip.close(); |
| } |
| } |
| |
| /** |
| * Find a jar that contains a class of the same name, if any. It will return a jar file, even if |
| * that is not the first thing on the class path that has a class with the same name. Looks first |
| * on the classpath and then in the <code>packagedClasses</code> map. |
| * @param my_class the class to find. |
| * @return a jar file that contains the class, or null. |
| */ |
| private static String findContainingJar(Class<?> my_class, Map<String, String> packagedClasses) |
| throws IOException { |
| ClassLoader loader = my_class.getClassLoader(); |
| |
| String class_file = my_class.getName().replaceAll("\\.", "/") + ".class"; |
| |
| if (loader != null) { |
| // first search the classpath |
| for (Enumeration<URL> itr = loader.getResources(class_file); itr.hasMoreElements();) { |
| URL url = itr.nextElement(); |
| if ("jar".equals(url.getProtocol())) { |
| String toReturn = url.getPath(); |
| if (toReturn.startsWith("file:")) { |
| toReturn = toReturn.substring("file:".length()); |
| } |
| // URLDecoder is a misnamed class, since it actually decodes |
| // x-www-form-urlencoded MIME type rather than actual |
| // URL encoding (which the file path has). Therefore it would |
| // decode +s to ' 's which is incorrect (spaces are actually |
| // either unencoded or encoded as "%20"). Replace +s first, so |
| // that they are kept sacred during the decoding process. |
| toReturn = toReturn.replaceAll("\\+", "%2B"); |
| toReturn = URLDecoder.decode(toReturn, "UTF-8"); |
| return toReturn.replaceAll("!.*$", ""); |
| } |
| } |
| } |
| |
| // now look in any jars we've packaged using JarFinder. Returns null when |
| // no jar is found. |
| return packagedClasses.get(class_file); |
| } |
| |
| /** |
| * Invoke 'getJar' on a custom JarFinder implementation. Useful for some job configuration |
| * contexts (HBASE-8140) and also for testing on MRv2. check if we have HADOOP-9426. |
| * @param my_class the class to find. |
| * @return a jar file that contains the class, or null. |
| */ |
| private static String getJar(Class<?> my_class) { |
| String ret = null; |
| try { |
| ret = JarFinder.getJar(my_class); |
| } catch (Exception e) { |
| // toss all other exceptions, related to reflection failure |
| throw new RuntimeException("getJar invocation failed.", e); |
| } |
| |
| return ret; |
| } |
| |
| private static int getRegionCount(Configuration conf, TableName tableName) throws IOException { |
| try (Connection conn = ConnectionFactory.createConnection(conf); |
| RegionLocator locator = conn.getRegionLocator(tableName)) { |
| return locator.getAllRegionLocations().size(); |
| } |
| } |
| } |
