HBASE-12247 Replace setHTable() with initializeTable() in TableInputFormat. (Solomon Duskis)
This commit is contained in:
parent
dc86001523
commit
3544f4e98b
|
@ -30,7 +30,6 @@ import org.apache.hadoop.hbase.classification.InterfaceAudience;
|
||||||
import org.apache.hadoop.hbase.classification.InterfaceStability;
|
import org.apache.hadoop.hbase.classification.InterfaceStability;
|
||||||
import org.apache.hadoop.hbase.client.Connection;
|
import org.apache.hadoop.hbase.client.Connection;
|
||||||
import org.apache.hadoop.hbase.client.ConnectionFactory;
|
import org.apache.hadoop.hbase.client.ConnectionFactory;
|
||||||
import org.apache.hadoop.hbase.client.HTable;
|
|
||||||
import org.apache.hadoop.hbase.client.RegionLocator;
|
import org.apache.hadoop.hbase.client.RegionLocator;
|
||||||
import org.apache.hadoop.hbase.client.Scan;
|
import org.apache.hadoop.hbase.client.Scan;
|
||||||
import org.apache.hadoop.hbase.util.Bytes;
|
import org.apache.hadoop.hbase.util.Bytes;
|
||||||
|
@ -109,9 +108,8 @@ implements Configurable {
|
||||||
this.conf = configuration;
|
this.conf = configuration;
|
||||||
TableName tableName = TableName.valueOf(conf.get(INPUT_TABLE));
|
TableName tableName = TableName.valueOf(conf.get(INPUT_TABLE));
|
||||||
try {
|
try {
|
||||||
// TODO: Replace setHTable() with initializeTable() once we have
|
// NOTE: This connection doesn't currently get closed explicit1ly.
|
||||||
// a clean method of closing a connection.
|
initializeTable(ConnectionFactory.createConnection(new Configuration(conf)), tableName);
|
||||||
setHTable(new HTable(new Configuration(conf), tableName));
|
|
||||||
} catch (Exception e) {
|
} catch (Exception e) {
|
||||||
LOG.error(StringUtils.stringifyException(e));
|
LOG.error(StringUtils.stringifyException(e));
|
||||||
}
|
}
|
||||||
|
@ -183,7 +181,6 @@ implements Configurable {
|
||||||
*
|
*
|
||||||
* @param scan The Scan to update.
|
* @param scan The Scan to update.
|
||||||
* @param familyAndQualifier family and qualifier
|
* @param familyAndQualifier family and qualifier
|
||||||
* @return A reference to this instance.
|
|
||||||
* @throws IllegalArgumentException When familyAndQualifier is invalid.
|
* @throws IllegalArgumentException When familyAndQualifier is invalid.
|
||||||
*/
|
*/
|
||||||
private static void addColumn(Scan scan, byte[] familyAndQualifier) {
|
private static void addColumn(Scan scan, byte[] familyAndQualifier) {
|
||||||
|
|
|
@ -96,6 +96,8 @@ extends InputFormat<ImmutableBytesWritable, Result> {
|
||||||
*
|
*
|
||||||
* @see Scan */
|
* @see Scan */
|
||||||
private Scan scan = null;
|
private Scan scan = null;
|
||||||
|
/** The {@link Admin}. */
|
||||||
|
private Admin admin;
|
||||||
/** The {@link Table} to scan. */
|
/** The {@link Table} to scan. */
|
||||||
private Table table;
|
private Table table;
|
||||||
/** The {@link RegionLocator} of the table. */
|
/** The {@link RegionLocator} of the table. */
|
||||||
|
@ -163,8 +165,8 @@ extends InputFormat<ImmutableBytesWritable, Result> {
|
||||||
throw new IOException("No table was provided.");
|
throw new IOException("No table was provided.");
|
||||||
}
|
}
|
||||||
|
|
||||||
RegionSizeCalculator sizeCalculator = new RegionSizeCalculator((HTable) table);
|
RegionSizeCalculator sizeCalculator = new RegionSizeCalculator(regionLocator, admin);
|
||||||
|
|
||||||
Pair<byte[][], byte[][]> keys = getStartEndKeys();
|
Pair<byte[][], byte[][]> keys = getStartEndKeys();
|
||||||
if (keys == null || keys.getFirst() == null ||
|
if (keys == null || keys.getFirst() == null ||
|
||||||
keys.getFirst().length == 0) {
|
keys.getFirst().length == 0) {
|
||||||
|
@ -287,25 +289,27 @@ extends InputFormat<ImmutableBytesWritable, Result> {
|
||||||
* Allows subclasses to set the {@link HTable}.
|
* Allows subclasses to set the {@link HTable}.
|
||||||
*
|
*
|
||||||
* @param table The table to get the data from.
|
* @param table The table to get the data from.
|
||||||
* @throws IOExceptfion
|
* @throws IOException
|
||||||
* @deprecated Use {@link #initializeTable(Connection, TableName)} instead.
|
* @deprecated Use {@link #initializeTable(Connection, TableName)} instead.
|
||||||
*/
|
*/
|
||||||
@Deprecated
|
@Deprecated
|
||||||
protected void setHTable(HTable table) throws IOException {
|
protected void setHTable(HTable table) throws IOException {
|
||||||
this.table = table;
|
this.table = table;
|
||||||
this.regionLocator = table;
|
this.regionLocator = table;
|
||||||
|
this.admin = table.getConnection().getAdmin();
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Allows subclasses to initalize the table information.
|
* Allows subclasses to initialize the table information.
|
||||||
*
|
*
|
||||||
* @param connection The {@link Connection} to the HBase cluster.
|
* @param connection The {@link Connection} to the HBase cluster.
|
||||||
* @param tableName The {@link TableName} of the table to process.
|
* @param tableName The {@link TableName} of the table to process.
|
||||||
* @throws IOExceptfion
|
* @throws IOException
|
||||||
*/
|
*/
|
||||||
protected void initializeTable(Connection connection, TableName tableName) throws IOException {
|
protected void initializeTable(Connection connection, TableName tableName) throws IOException {
|
||||||
this.table = connection.getTable(tableName);
|
this.table = connection.getTable(tableName);
|
||||||
this.regionLocator = connection.getRegionLocator(tableName);
|
this.regionLocator = connection.getRegionLocator(tableName);
|
||||||
|
this.admin = connection.getAdmin();
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
Loading…
Reference in New Issue