org.apache.hadoop.hbase.mapreduce

Class TableMapReduceUtil

java.lang.Object

org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil

@InterfaceAudience.Public
 @InterfaceStability.Stable
public class TableMapReduceUtil
extends Object

Utility for TableMapper and TableReducer

Constructor Summary

Constructors

Constructor and Description
TableMapReduceUtil()

Method Summary

Modifier and Type	Method and Description
static void	addDependencyJars(Configuration conf, Class<?>... classes) 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.
static void	addDependencyJars(Job job) 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.
static void	addHBaseDependencyJars(Configuration conf) Add HBase and its dependencies (only) to the job configuration.
static String	buildDependencyClasspath(Configuration conf) Returns a classpath string built from the content of the "tmpjars" value in conf.
static void	initCredentials(Job job)
static void	initCredentialsForCluster(Job job, Configuration conf) 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.
static void	initCredentialsForCluster(Job job, String quorumAddress) Deprecated. Since 1.2.0, use initCredentialsForCluster(Job, Configuration) instead.
static void	initTableMapperJob(byte[] table, Scan scan, Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, Job job) Use this before submitting a TableMap job.
static void	initTableMapperJob(byte[] table, Scan scan, Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, Job job, boolean addDependencyJars) Use this before submitting a TableMap job.
static void	initTableMapperJob(byte[] table, Scan scan, Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, Job job, boolean addDependencyJars, Class<? extends InputFormat> inputFormatClass) Use this before submitting a TableMap job.
static void	initTableMapperJob(List<Scan> scans, Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, Job job) Use this before submitting a Multi TableMap job.
static void	initTableMapperJob(List<Scan> scans, Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, Job job, boolean addDependencyJars) Use this before submitting a Multi TableMap job.
static void	initTableMapperJob(List<Scan> scans, Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, Job job, boolean addDependencyJars, boolean initCredentials) Use this before submitting a Multi TableMap job.
static void	initTableMapperJob(String table, Scan scan, Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, Job job) Use this before submitting a TableMap job.
static void	initTableMapperJob(String table, Scan scan, Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, Job job, boolean addDependencyJars) Use this before submitting a TableMap job.
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) Use this before submitting a TableMap job.
static void	initTableMapperJob(String table, Scan scan, Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, Job job, boolean addDependencyJars, Class<? extends InputFormat> inputFormatClass) Use this before submitting a TableMap job.
static void	initTableMapperJob(TableName table, Scan scan, Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, Job job) Use this before submitting a TableMap job.
static void	initTableReducerJob(String table, Class<? extends TableReducer> reducer, Job job) Use this before submitting a TableReduce job.
static void	initTableReducerJob(String table, Class<? extends TableReducer> reducer, Job job, Class partitioner) Use this before submitting a TableReduce job.
static void	initTableReducerJob(String table, Class<? extends TableReducer> reducer, Job job, Class partitioner, String quorumAddress, String serverClass, String serverImpl) Use this before submitting a TableReduce job.
static void	initTableReducerJob(String table, Class<? extends TableReducer> reducer, Job job, Class partitioner, String quorumAddress, String serverClass, String serverImpl, boolean addDependencyJars) Use this before submitting a TableReduce job.
static void	initTableSnapshotMapperJob(String snapshotName, Scan scan, Class<? extends TableMapper> mapper, Class<?> outputKeyClass, Class<?> outputValueClass, Job job, boolean addDependencyJars, Path tmpRestoreDir) Sets up the job for reading from a table snapshot.
static void	limitNumReduceTasks(String table, Job 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.
static void	resetCacheConfig(Configuration conf) Enable a basic on-heap cache for these jobs.
static void	setNumReduceTasks(String table, Job job) Sets the number of reduce tasks for the given job configuration to the number of regions the given table has.
static void	setScannerCaching(Job job, int batchSize) Sets the number of rows to return and cache with each scanner iteration.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

TableMapReduceUtil

public TableMapReduceUtil()

Method Detail

initTableMapperJob

public static void initTableMapperJob(String table,
                                      Scan scan,
                                      Class<? extends TableMapper> mapper,
                                      Class<?> outputKeyClass,
                                      Class<?> outputValueClass,
                                      Job job)
                               throws IOException

Use this before submitting a TableMap job. It will appropriately set up the job.

Parameters:

table - The table name to read from.

scan - The scan instance with the columns, time range etc.

mapper - The mapper class to use.

outputKeyClass - The class of the output key.

outputValueClass - The class of the output value.

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.

initTableMapperJob

public static void initTableMapperJob(TableName table,
                                      Scan scan,
                                      Class<? extends TableMapper> mapper,
                                      Class<?> outputKeyClass,
                                      Class<?> outputValueClass,
                                      Job job)
                               throws IOException

Use this before submitting a TableMap job. It will appropriately set up the job.

Parameters:

table - The table name to read from.

scan - The scan instance with the columns, time range etc.

mapper - The mapper class to use.

outputKeyClass - The class of the output key.

outputValueClass - The class of the output value.

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.

initTableMapperJob

public static void initTableMapperJob(byte[] table,
                                      Scan scan,
                                      Class<? extends TableMapper> mapper,
                                      Class<?> outputKeyClass,
                                      Class<?> outputValueClass,
                                      Job job)
                               throws IOException

Use this before submitting a TableMap job. It will appropriately set up the job.

Parameters:

table - Binary representation of the table name to read from.

scan - The scan instance with the columns, time range etc.

mapper - The mapper class to use.

outputKeyClass - The class of the output key.

outputValueClass - The class of the output value.

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.

initTableMapperJob

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

Use this before submitting a TableMap job. It will appropriately set up the job.

Parameters:

table - The table name to read from.

scan - The scan instance with the columns, time range etc.

mapper - The mapper class to use.

outputKeyClass - The class of the output key.

outputValueClass - The class of the output value.

job - The current job to adjust. Make sure the passed job is carrying all necessary HBase configuration.

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.

initTableMapperJob

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

Use this before submitting a TableMap job. It will appropriately set up the job.

Parameters:

table - The table name to read from.

scan - The scan instance with the columns, time range etc.

mapper - The mapper class to use.

outputKeyClass - The class of the output key.

outputValueClass - The class of the output value.

job - The current job to adjust. Make sure the passed job is carrying all necessary HBase configuration.

addDependencyJars - upload HBase jars and jars for any of the configured job classes via the distributed cache (tmpjars).

initCredentials - whether to initialize hbase auth credentials for the job

inputFormatClass - the input format

Throws:

IOException - When setting up the details fails.

initTableMapperJob

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

Use this before submitting a TableMap job. It will appropriately set up the job.

Parameters:

table - Binary representation of the table name to read from.

scan - The scan instance with the columns, time range etc.

mapper - The mapper class to use.

outputKeyClass - The class of the output key.

outputValueClass - The class of the output value.

job - The current job to adjust. Make sure the passed job is carrying all necessary HBase configuration.

addDependencyJars - upload HBase jars and jars for any of the configured job classes via the distributed cache (tmpjars).

inputFormatClass - The class of the input format

Throws:

IOException - When setting up the details fails.

initTableMapperJob

public static void initTableMapperJob(byte[] table,
                                      Scan scan,
                                      Class<? extends TableMapper> mapper,
                                      Class<?> outputKeyClass,
                                      Class<?> outputValueClass,
                                      Job job,
                                      boolean addDependencyJars)
                               throws IOException

Use this before submitting a TableMap job. It will appropriately set up the job.

Parameters:

table - Binary representation of the table name to read from.

scan - The scan instance with the columns, time range etc.

mapper - The mapper class to use.

outputKeyClass - The class of the output key.

outputValueClass - The class of the output value.

job - The current job to adjust. Make sure the passed job is carrying all necessary HBase configuration.

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.

initTableMapperJob

public static void initTableMapperJob(String table,
                                      Scan scan,
                                      Class<? extends TableMapper> mapper,
                                      Class<?> outputKeyClass,
                                      Class<?> outputValueClass,
                                      Job job,
                                      boolean addDependencyJars)
                               throws IOException

Use this before submitting a TableMap job. It will appropriately set up the job.

Parameters:

table - The table name to read from.

scan - The scan instance with the columns, time range etc.

mapper - The mapper class to use.

outputKeyClass - The class of the output key.

outputValueClass - The class of the output value.

job - The current job to adjust. Make sure the passed job is carrying all necessary HBase configuration.

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.

resetCacheConfig

public static void resetCacheConfig(Configuration conf)

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.

initTableSnapshotMapperJob

public static void initTableSnapshotMapperJob(String snapshotName,
                                              Scan scan,
                                              Class<? extends TableMapper> mapper,
                                              Class<?> outputKeyClass,
                                              Class<?> outputValueClass,
                                              Job job,
                                              boolean addDependencyJars,
                                              Path tmpRestoreDir)
                                       throws IOException

Sets up the job for reading from a table snapshot. It bypasses hbase servers and read directly from snapshot files.

Parameters:

snapshotName - The name of the snapshot (of a table) to read from.

scan - The scan instance with the columns, time range etc.

mapper - The mapper class to use.

outputKeyClass - The class of the output key.

outputValueClass - The class of the output value.

job - The current job to adjust. Make sure the passed job is carrying all necessary HBase configuration.

addDependencyJars - upload HBase jars and jars for any of the configured job classes via the distributed cache (tmpjars).

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 Also:

TableSnapshotInputFormat

initTableMapperJob

public static void initTableMapperJob(List<Scan> scans,
                                      Class<? extends TableMapper> mapper,
                                      Class<?> outputKeyClass,
                                      Class<?> outputValueClass,
                                      Job job)
                               throws IOException

Use this before submitting a Multi TableMap job. It will appropriately set up the job.

Parameters:

scans - The list of Scan objects to read from.

mapper - The mapper class to use.

outputKeyClass - The class of the output key.

outputValueClass - The class of the output value.

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.

initTableMapperJob

public static void initTableMapperJob(List<Scan> scans,
                                      Class<? extends TableMapper> mapper,
                                      Class<?> outputKeyClass,
                                      Class<?> outputValueClass,
                                      Job job,
                                      boolean addDependencyJars)
                               throws IOException

Use this before submitting a Multi TableMap job. It will appropriately set up the job.

Parameters:

scans - The list of Scan objects to read from.

mapper - The mapper class to use.

outputKeyClass - The class of the output key.

outputValueClass - The class of the output value.

job - The current job to adjust. Make sure the passed job is carrying all necessary HBase configuration.

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.

initTableMapperJob

public static void initTableMapperJob(List<Scan> scans,
                                      Class<? extends TableMapper> mapper,
                                      Class<?> outputKeyClass,
                                      Class<?> outputValueClass,
                                      Job job,
                                      boolean addDependencyJars,
                                      boolean initCredentials)
                               throws IOException

Use this before submitting a Multi TableMap job. It will appropriately set up the job.

Parameters:

scans - The list of Scan objects to read from.

mapper - The mapper class to use.

outputKeyClass - The class of the output key.

outputValueClass - The class of the output value.

job - The current job to adjust. Make sure the passed job is carrying all necessary HBase configuration.

addDependencyJars - upload HBase jars and jars for any of the configured job classes via the distributed cache (tmpjars).

initCredentials - whether to initialize hbase auth credentials for the job

Throws:

IOException - When setting up the details fails.

initCredentials

public static void initCredentials(Job job)
                            throws IOException

Throws:

IOException

initCredentialsForCluster

@Deprecated
public static void initCredentialsForCluster(Job job,
                                                         String quorumAddress)
                                                  throws IOException

Deprecated. Since 1.2.0, use initCredentialsForCluster(Job, Configuration) instead.

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. The quorumAddress is the key to the ZK ensemble, which contains: hbase.zookeeper.quorum, hbase.zookeeper.client.port and zookeeper.znode.parent

Parameters:

job - The job that requires the permission.

quorumAddress - string that contains the 3 required configuratins

Throws:

IOException - When the authentication token cannot be obtained.

initCredentialsForCluster

public static void initCredentialsForCluster(Job job,
                                             Configuration conf)
                                      throws IOException

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.

Parameters:

job - The job that requires the permission.

conf - The configuration to use in connecting to the peer cluster

Throws:

IOException - When the authentication token cannot be obtained.

initTableReducerJob

public static void initTableReducerJob(String table,
                                       Class<? extends TableReducer> reducer,
                                       Job job)
                                throws IOException

Use this before submitting a TableReduce job. It will appropriately set up the JobConf.

Parameters:

table - The output table.

reducer - The reducer class to use.

job - The current job to adjust.

Throws:

IOException - When determining the region count fails.

initTableReducerJob

public static void initTableReducerJob(String table,
                                       Class<? extends TableReducer> reducer,
                                       Job job,
                                       Class partitioner)
                                throws IOException

Use this before submitting a TableReduce job. It will appropriately set up the JobConf.

Parameters:

table - The output table.

reducer - The reducer class to use.

job - The current job to adjust.

partitioner - Partitioner to use. Pass null to use default partitioner.

Throws:

IOException - When determining the region count fails.

initTableReducerJob

public static void initTableReducerJob(String table,
                                       Class<? extends TableReducer> reducer,
                                       Job job,
                                       Class partitioner,
                                       String quorumAddress,
                                       String serverClass,
                                       String serverImpl)
                                throws IOException

Use this before submitting a TableReduce job. It will appropriately set up the JobConf.

Parameters:

table - The output table.

reducer - The reducer class to use.

job - The current job to adjust. Make sure the passed job is carrying all necessary HBase configuration.

partitioner - Partitioner to use. Pass null to use default partitioner.

quorumAddress - Distant cluster to write to; default is null for output to the cluster that is designated in hbase-site.xml. 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 hbase-site.xml and this param would have the ensemble address of the remote cluster. The format to pass is particular. Pass <hbase.zookeeper.quorum>:<hbase.zookeeper.client.port>:<zookeeper.znode.parent> such as server,server2,server3:2181:/hbase.

serverClass - redefined hbase.regionserver.class

serverImpl - redefined hbase.regionserver.impl

Throws:

IOException - When determining the region count fails.

initTableReducerJob

public static void initTableReducerJob(String table,
                                       Class<? extends TableReducer> reducer,
                                       Job job,
                                       Class partitioner,
                                       String quorumAddress,
                                       String serverClass,
                                       String serverImpl,
                                       boolean addDependencyJars)
                                throws IOException

Use this before submitting a TableReduce job. It will appropriately set up the JobConf.

Parameters:

table - The output table.

reducer - The reducer class to use.

job - The current job to adjust. Make sure the passed job is carrying all necessary HBase configuration.

partitioner - Partitioner to use. Pass null to use default partitioner.

quorumAddress - Distant cluster to write to; default is null for output to the cluster that is designated in hbase-site.xml. 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 hbase-site.xml and this param would have the ensemble address of the remote cluster. The format to pass is particular. Pass <hbase.zookeeper.quorum>:<hbase.zookeeper.client.port>:<zookeeper.znode.parent> such as server,server2,server3:2181:/hbase.

serverClass - redefined hbase.regionserver.class

serverImpl - redefined hbase.regionserver.impl

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.

limitNumReduceTasks

public static void limitNumReduceTasks(String table,
                                       Job job)
                                throws IOException

Ensures that the given number of reduce tasks for the given job configuration does not exceed the number of regions for the given table.

Parameters:

table - The table to get the region count for.

job - The current job to adjust.

Throws:

IOException - When retrieving the table details fails.

setNumReduceTasks

public static void setNumReduceTasks(String table,
                                     Job job)
                              throws IOException

Sets the number of reduce tasks for the given job configuration to the number of regions the given table has.

Parameters:

table - The table to get the region count for.

job - The current job to adjust.

Throws:

IOException - When retrieving the table details fails.

setScannerCaching

public static void setScannerCaching(Job job,
                                     int batchSize)

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.

Parameters:

job - The current job to adjust.

batchSize - The number of rows to return in batch with each scanner iteration.

addHBaseDependencyJars

public static void addHBaseDependencyJars(Configuration conf)
                                   throws IOException

Add HBase and its dependencies (only) to the job configuration.

This is intended as a low-level API, facilitating code reuse between this class and its mapred counterpart. It also of use to extenral tools that need to build a MapReduce job that interacts with HBase but want fine-grained control over the jars shipped to the cluster.

Parameters:

conf - The Configuration object to extend with dependencies.

Throws:

IOException

See Also:

TableMapReduceUtil, PIG-3285

buildDependencyClasspath

public static String buildDependencyClasspath(Configuration conf)

Returns a classpath string built from the content of the "tmpjars" value in conf. Also exposed to shell scripts via `bin/hbase mapredcp`.

addDependencyJars

public static void addDependencyJars(Job job)
                              throws IOException

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.

Throws:

IOException

addDependencyJars

public static void addDependencyJars(Configuration conf,
                                     Class<?>... classes)
                              throws IOException

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.

Throws:

IOException