- 浏览: 94527 次
- 性别:
- 来自: 北京
最新评论
-
leibnitz:
这么早就研究源码已经是难得了.你说的第二点能这样说吗:a.会依 ...
Hbase的Region Compact算法实现分析 -
uestzengting:
朝阳之辉 写道你的想法是保证rowkey唯一是吧,我的想法是r ...
HBase bulkload的一个bug定位 -
朝阳之辉:
你的想法是保证rowkey唯一是吧,我的想法是rowkey是可 ...
HBase bulkload的一个bug定位 -
uestzengting:
朝阳之辉 写道你的version是用的时间戳吧,在数据量非常大 ...
HBase bulkload的一个bug定位 -
朝阳之辉:
你的version是用的时间戳吧,在数据量非常大的情况下,怎样 ...
HBase bulkload的一个bug定位
随着Hbase里删除的进行,有些Region的数据会越来越少,而HBase不会主动去回收这些Region,因此会造成Region越来越多。HBase里提供了一个工具类HMerge,直接拿过来用却并不能运行,按照自已对HBase的理解对HMerge稍做修改,使其可以运行。运行时不需要Disable表,但需要注意的是在运行时如果同时往该表里put数据,可能会有问题,改后的代码如下:
/**
* Copyright 2009 The Apache Software Foundation
*
* 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 hbase;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.RemoteExceptionHandler;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.regionserver.HRegion;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.regionserver.wal.HLog;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Writables;
/**
* A non-instantiable class that has a static method capable of compacting
* a table by merging adjacent regions.
*/
public class HMerge {
static final Log LOG = LogFactory.getLog(HMerge.class);
static final Random rand = new Random();
/*
* Not instantiable
*/
private HMerge() {
super();
}
/**
* Scans the table and merges two adjacent regions if they are small. This
* only happens when a lot of rows are deleted.
*
* When merging the META region, the HBase instance must be offline.
* When merging a normal table, the HBase instance must be online, but the
* table must be disabled.
*
* @param conf - configuration object for HBase
* @param fs - FileSystem where regions reside
* @param tableName - Table to be compacted
* @throws IOException
*/
public static void merge(Configuration conf, FileSystem fs, final byte[] tableName)
throws IOException {
merge(conf, fs, tableName, true);
}
/**
* Scans the table and merges two adjacent regions if they are small. This
* only happens when a lot of rows are deleted.
*
* When merging the META region, the HBase instance must be offline.
* When merging a normal table, the HBase instance must be online, but the
* table must be disabled.
*
* @param conf - configuration object for HBase
* @param fs - FileSystem where regions reside
* @param tableName - Table to be compacted
* @param testMasterRunning True if we are to verify master is down before
* running merge
* @throws IOException
*/
public static void merge(Configuration conf, FileSystem fs, final byte[] tableName,
final boolean testMasterRunning) throws IOException {
boolean masterIsRunning = false;
if (testMasterRunning) {
HConnection connection = HConnectionManager.getConnection(conf);
masterIsRunning = connection.isMasterRunning();
}
HConnectionManager.deleteConnection(conf, true);
if (Bytes.equals(tableName, HConstants.META_TABLE_NAME)) {
if (masterIsRunning) {
throw new IllegalStateException("Can not compact META table if instance is on-line");
}
new OfflineMerger(conf, fs).process();
} else {
if (!masterIsRunning) {
throw new IllegalStateException(
"HBase instance must be running to merge a normal table");
}
// HBaseAdmin admin = new HBaseAdmin(conf);
// if (!admin.isTableDisabled(tableName)) {
// throw new TableNotDisabledException(tableName);
// }
new OnlineMerger(conf, fs, tableName).process();
}
}
private static abstract class Merger {
protected final Configuration conf;
protected final FileSystem fs;
protected final Path tabledir;
protected final HLog hlog;
private final long maxFilesize;
protected Merger(Configuration conf, FileSystem fs, final byte[] tableName)
throws IOException {
this.conf = conf;
this.fs = fs;
this.maxFilesize = conf.getLong("hbase.hregion.max.filesize",
HConstants.DEFAULT_MAX_FILE_SIZE);
this.tabledir = new Path(fs.makeQualified(new Path(conf.get(HConstants.HBASE_DIR))),
Bytes.toString(tableName));
Path logdir = new Path(tabledir, "merge_" + System.currentTimeMillis()
+ HConstants.HREGION_LOGDIR_NAME);
Path oldLogDir = new Path(tabledir, HConstants.HREGION_OLDLOGDIR_NAME);
this.hlog = new HLog(fs, logdir, oldLogDir, conf);
}
void process() throws IOException {
try {
for (HRegionInfo[] regionsToMerge = next(); regionsToMerge != null; regionsToMerge = next()) {
if (!merge(regionsToMerge)) {
return;
}
}
} finally {
try {
hlog.closeAndDelete();
} catch (IOException e) {
LOG.error(e);
}
}
}
protected boolean merge(final HRegionInfo[] info) throws IOException {
if (info.length < 2) {
LOG.info("only one region - nothing to merge");
return false;
}
HRegion currentRegion = null;
long currentSize = 0;
HRegion nextRegion = null;
long nextSize = 0;
for (int i = 0; i < info.length - 1; i++) {
if (currentRegion == null) {
currentRegion = HRegion.newHRegion(tabledir, hlog, fs, conf, info[i], null);
currentRegion.initialize();
currentSize = currentRegion.getLargestHStoreSize();
}
nextRegion = HRegion.newHRegion(tabledir, hlog, fs, conf, info[i + 1], null);
nextRegion.initialize();
nextSize = nextRegion.getLargestHStoreSize();
if ((currentSize + nextSize) <= (maxFilesize / 2)) {
// We merge two adjacent regions if their total size is less than
// one half of the desired maximum size
LOG.info("Merging regions " + currentRegion.getRegionNameAsString() + " and "
+ nextRegion.getRegionNameAsString());
HRegion mergedRegion = HRegion.mergeAdjacent(currentRegion, nextRegion);
updateMeta(currentRegion.getRegionName(), nextRegion.getRegionName(),
mergedRegion);
break;
}
LOG.info("not merging regions " + Bytes.toString(currentRegion.getRegionName())
+ " and " + Bytes.toString(nextRegion.getRegionName()));
currentRegion.close();
currentRegion = nextRegion;
currentSize = nextSize;
}
if (currentRegion != null) {
currentRegion.close();
}
return true;
}
protected abstract HRegionInfo[] next() throws IOException;
protected abstract void updateMeta(final byte[] oldRegion1, final byte[] oldRegion2,
HRegion newRegion) throws IOException;
}
/** Instantiated to compact a normal user table */
private static class OnlineMerger extends Merger {
private final byte[] tableName;
private final HTable table;
private final ResultScanner metaScanner;
private HRegionInfo latestRegion;
OnlineMerger(Configuration conf, FileSystem fs, final byte[] tableName) throws IOException {
super(conf, fs, tableName);
this.tableName = tableName;
this.table = new HTable(conf, HConstants.META_TABLE_NAME);
this.metaScanner = table.getScanner(HConstants.CATALOG_FAMILY,
HConstants.REGIONINFO_QUALIFIER);
this.latestRegion = null;
}
private HRegionInfo nextRegion() throws IOException {
try {
HRegionInfo results = getMetaRow();
return results;
} catch (IOException e) {
e = RemoteExceptionHandler.checkIOException(e);
LOG.error("meta scanner error", e);
metaScanner.close();
throw e;
}
}
/*
* Check current row has a HRegionInfo. Skip to next row if HRI is empty.
* @return A Map of the row content else null if we are off the end.
* @throws IOException
*/
private HRegionInfo getMetaRow() throws IOException {
Result currentRow = metaScanner.next();
boolean found = false;
HRegionInfo region = null;
while (currentRow != null) {
LOG.info("Row: <" + Bytes.toString(currentRow.getRow()) + ">");
byte[] regionInfoValue = currentRow.getValue(HConstants.CATALOG_FAMILY,
HConstants.REGIONINFO_QUALIFIER);
if (regionInfoValue == null || regionInfoValue.length == 0) {
currentRow = metaScanner.next();
continue;
} else {
region = Writables.getHRegionInfo(regionInfoValue);
if (!Bytes.equals(region.getTableDesc().getName(), this.tableName)) {
currentRow = metaScanner.next();
continue;
}
}
found = true;
break;
}
return found ? region : null;
}
@Override
protected HRegionInfo[] next() throws IOException {
List<HRegionInfo> regions = new ArrayList<HRegionInfo>();
if (latestRegion == null) {
latestRegion = nextRegion();
}
if (latestRegion != null) {
regions.add(latestRegion);
}
latestRegion = nextRegion();
if (latestRegion != null) {
regions.add(latestRegion);
}
return regions.toArray(new HRegionInfo[regions.size()]);
}
@Override
protected void updateMeta(final byte[] oldRegion1, final byte[] oldRegion2,
HRegion newRegion) throws IOException {
byte[][] regionsToDelete = { oldRegion1, oldRegion2 };
for (int r = 0; r < regionsToDelete.length; r++) {
if (Bytes.equals(regionsToDelete[r], latestRegion.getRegionName())) {
latestRegion = null;
}
Delete delete = new Delete(regionsToDelete[r]);
table.delete(delete);
LOG.info("updated columns in row: " + Bytes.toString(regionsToDelete[r]));
}
Put put = new Put(newRegion.getRegionName());
put.add(HConstants.CATALOG_FAMILY, HConstants.REGIONINFO_QUALIFIER, Writables
.getBytes(newRegion.getRegionInfo()));
table.put(put);
HBaseAdmin admin = new HBaseAdmin(conf);
admin.getMaster().assign(newRegion.getRegionName(), true);
LOG.info("updated columns in row: " + Bytes.toString(newRegion.getRegionName()));
}
}
/** Instantiated to compact the meta region */
private static class OfflineMerger extends Merger {
private final List<HRegionInfo> metaRegions = new ArrayList<HRegionInfo>();
private final HRegion root;
OfflineMerger(Configuration conf, FileSystem fs) throws IOException {
super(conf, fs, HConstants.META_TABLE_NAME);
Path rootTableDir = HTableDescriptor.getTableDir(fs.makeQualified(new Path(conf
.get(HConstants.HBASE_DIR))), HConstants.ROOT_TABLE_NAME);
// Scan root region to find all the meta regions
root = HRegion.newHRegion(rootTableDir, hlog, fs, conf, HRegionInfo.ROOT_REGIONINFO,
null);
root.initialize();
Scan scan = new Scan();
scan.addColumn(HConstants.CATALOG_FAMILY, HConstants.REGIONINFO_QUALIFIER);
InternalScanner rootScanner = root.getScanner(scan);
try {
List<KeyValue> results = new ArrayList<KeyValue>();
while (rootScanner.next(results)) {
for (KeyValue kv : results) {
HRegionInfo info = Writables.getHRegionInfoOrNull(kv.getValue());
if (info != null) {
metaRegions.add(info);
}
}
}
} finally {
rootScanner.close();
try {
root.close();
} catch (IOException e) {
LOG.error(e);
}
}
}
@Override
protected HRegionInfo[] next() {
HRegionInfo[] results = null;
if (metaRegions.size() > 0) {
results = metaRegions.toArray(new HRegionInfo[metaRegions.size()]);
metaRegions.clear();
}
return results;
}
@Override
protected void updateMeta(final byte[] oldRegion1, final byte[] oldRegion2,
HRegion newRegion) throws IOException {
byte[][] regionsToDelete = { oldRegion1, oldRegion2 };
for (int r = 0; r < regionsToDelete.length; r++) {
Delete delete = new Delete(regionsToDelete[r]);
delete.deleteColumns(HConstants.CATALOG_FAMILY, HConstants.REGIONINFO_QUALIFIER);
delete.deleteColumns(HConstants.CATALOG_FAMILY, HConstants.SERVER_QUALIFIER);
delete.deleteColumns(HConstants.CATALOG_FAMILY, HConstants.STARTCODE_QUALIFIER);
delete.deleteColumns(HConstants.CATALOG_FAMILY, HConstants.SPLITA_QUALIFIER);
delete.deleteColumns(HConstants.CATALOG_FAMILY, HConstants.SPLITB_QUALIFIER);
root.delete(delete, null, true);
LOG.info("updated columns in row: " + Bytes.toString(regionsToDelete[r]));
}
HRegionInfo newInfo = newRegion.getRegionInfo();
newInfo.setOffline(true);
Put put = new Put(newRegion.getRegionName());
put.add(HConstants.CATALOG_FAMILY, HConstants.REGIONINFO_QUALIFIER, Writables
.getBytes(newInfo));
root.put(put);
LOG.info("updated columns in row: " + Bytes.toString(newRegion.getRegionName()));
}
}
}
/**
* Copyright 2009 The Apache Software Foundation
*
* 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 hbase;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.RemoteExceptionHandler;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.regionserver.HRegion;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.regionserver.wal.HLog;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Writables;
/**
* A non-instantiable class that has a static method capable of compacting
* a table by merging adjacent regions.
*/
public class HMerge {
static final Log LOG = LogFactory.getLog(HMerge.class);
static final Random rand = new Random();
/*
* Not instantiable
*/
private HMerge() {
super();
}
/**
* Scans the table and merges two adjacent regions if they are small. This
* only happens when a lot of rows are deleted.
*
* When merging the META region, the HBase instance must be offline.
* When merging a normal table, the HBase instance must be online, but the
* table must be disabled.
*
* @param conf - configuration object for HBase
* @param fs - FileSystem where regions reside
* @param tableName - Table to be compacted
* @throws IOException
*/
public static void merge(Configuration conf, FileSystem fs, final byte[] tableName)
throws IOException {
merge(conf, fs, tableName, true);
}
/**
* Scans the table and merges two adjacent regions if they are small. This
* only happens when a lot of rows are deleted.
*
* When merging the META region, the HBase instance must be offline.
* When merging a normal table, the HBase instance must be online, but the
* table must be disabled.
*
* @param conf - configuration object for HBase
* @param fs - FileSystem where regions reside
* @param tableName - Table to be compacted
* @param testMasterRunning True if we are to verify master is down before
* running merge
* @throws IOException
*/
public static void merge(Configuration conf, FileSystem fs, final byte[] tableName,
final boolean testMasterRunning) throws IOException {
boolean masterIsRunning = false;
if (testMasterRunning) {
HConnection connection = HConnectionManager.getConnection(conf);
masterIsRunning = connection.isMasterRunning();
}
HConnectionManager.deleteConnection(conf, true);
if (Bytes.equals(tableName, HConstants.META_TABLE_NAME)) {
if (masterIsRunning) {
throw new IllegalStateException("Can not compact META table if instance is on-line");
}
new OfflineMerger(conf, fs).process();
} else {
if (!masterIsRunning) {
throw new IllegalStateException(
"HBase instance must be running to merge a normal table");
}
// HBaseAdmin admin = new HBaseAdmin(conf);
// if (!admin.isTableDisabled(tableName)) {
// throw new TableNotDisabledException(tableName);
// }
new OnlineMerger(conf, fs, tableName).process();
}
}
private static abstract class Merger {
protected final Configuration conf;
protected final FileSystem fs;
protected final Path tabledir;
protected final HLog hlog;
private final long maxFilesize;
protected Merger(Configuration conf, FileSystem fs, final byte[] tableName)
throws IOException {
this.conf = conf;
this.fs = fs;
this.maxFilesize = conf.getLong("hbase.hregion.max.filesize",
HConstants.DEFAULT_MAX_FILE_SIZE);
this.tabledir = new Path(fs.makeQualified(new Path(conf.get(HConstants.HBASE_DIR))),
Bytes.toString(tableName));
Path logdir = new Path(tabledir, "merge_" + System.currentTimeMillis()
+ HConstants.HREGION_LOGDIR_NAME);
Path oldLogDir = new Path(tabledir, HConstants.HREGION_OLDLOGDIR_NAME);
this.hlog = new HLog(fs, logdir, oldLogDir, conf);
}
void process() throws IOException {
try {
for (HRegionInfo[] regionsToMerge = next(); regionsToMerge != null; regionsToMerge = next()) {
if (!merge(regionsToMerge)) {
return;
}
}
} finally {
try {
hlog.closeAndDelete();
} catch (IOException e) {
LOG.error(e);
}
}
}
protected boolean merge(final HRegionInfo[] info) throws IOException {
if (info.length < 2) {
LOG.info("only one region - nothing to merge");
return false;
}
HRegion currentRegion = null;
long currentSize = 0;
HRegion nextRegion = null;
long nextSize = 0;
for (int i = 0; i < info.length - 1; i++) {
if (currentRegion == null) {
currentRegion = HRegion.newHRegion(tabledir, hlog, fs, conf, info[i], null);
currentRegion.initialize();
currentSize = currentRegion.getLargestHStoreSize();
}
nextRegion = HRegion.newHRegion(tabledir, hlog, fs, conf, info[i + 1], null);
nextRegion.initialize();
nextSize = nextRegion.getLargestHStoreSize();
if ((currentSize + nextSize) <= (maxFilesize / 2)) {
// We merge two adjacent regions if their total size is less than
// one half of the desired maximum size
LOG.info("Merging regions " + currentRegion.getRegionNameAsString() + " and "
+ nextRegion.getRegionNameAsString());
HRegion mergedRegion = HRegion.mergeAdjacent(currentRegion, nextRegion);
updateMeta(currentRegion.getRegionName(), nextRegion.getRegionName(),
mergedRegion);
break;
}
LOG.info("not merging regions " + Bytes.toString(currentRegion.getRegionName())
+ " and " + Bytes.toString(nextRegion.getRegionName()));
currentRegion.close();
currentRegion = nextRegion;
currentSize = nextSize;
}
if (currentRegion != null) {
currentRegion.close();
}
return true;
}
protected abstract HRegionInfo[] next() throws IOException;
protected abstract void updateMeta(final byte[] oldRegion1, final byte[] oldRegion2,
HRegion newRegion) throws IOException;
}
/** Instantiated to compact a normal user table */
private static class OnlineMerger extends Merger {
private final byte[] tableName;
private final HTable table;
private final ResultScanner metaScanner;
private HRegionInfo latestRegion;
OnlineMerger(Configuration conf, FileSystem fs, final byte[] tableName) throws IOException {
super(conf, fs, tableName);
this.tableName = tableName;
this.table = new HTable(conf, HConstants.META_TABLE_NAME);
this.metaScanner = table.getScanner(HConstants.CATALOG_FAMILY,
HConstants.REGIONINFO_QUALIFIER);
this.latestRegion = null;
}
private HRegionInfo nextRegion() throws IOException {
try {
HRegionInfo results = getMetaRow();
return results;
} catch (IOException e) {
e = RemoteExceptionHandler.checkIOException(e);
LOG.error("meta scanner error", e);
metaScanner.close();
throw e;
}
}
/*
* Check current row has a HRegionInfo. Skip to next row if HRI is empty.
* @return A Map of the row content else null if we are off the end.
* @throws IOException
*/
private HRegionInfo getMetaRow() throws IOException {
Result currentRow = metaScanner.next();
boolean found = false;
HRegionInfo region = null;
while (currentRow != null) {
LOG.info("Row: <" + Bytes.toString(currentRow.getRow()) + ">");
byte[] regionInfoValue = currentRow.getValue(HConstants.CATALOG_FAMILY,
HConstants.REGIONINFO_QUALIFIER);
if (regionInfoValue == null || regionInfoValue.length == 0) {
currentRow = metaScanner.next();
continue;
} else {
region = Writables.getHRegionInfo(regionInfoValue);
if (!Bytes.equals(region.getTableDesc().getName(), this.tableName)) {
currentRow = metaScanner.next();
continue;
}
}
found = true;
break;
}
return found ? region : null;
}
@Override
protected HRegionInfo[] next() throws IOException {
List<HRegionInfo> regions = new ArrayList<HRegionInfo>();
if (latestRegion == null) {
latestRegion = nextRegion();
}
if (latestRegion != null) {
regions.add(latestRegion);
}
latestRegion = nextRegion();
if (latestRegion != null) {
regions.add(latestRegion);
}
return regions.toArray(new HRegionInfo[regions.size()]);
}
@Override
protected void updateMeta(final byte[] oldRegion1, final byte[] oldRegion2,
HRegion newRegion) throws IOException {
byte[][] regionsToDelete = { oldRegion1, oldRegion2 };
for (int r = 0; r < regionsToDelete.length; r++) {
if (Bytes.equals(regionsToDelete[r], latestRegion.getRegionName())) {
latestRegion = null;
}
Delete delete = new Delete(regionsToDelete[r]);
table.delete(delete);
LOG.info("updated columns in row: " + Bytes.toString(regionsToDelete[r]));
}
Put put = new Put(newRegion.getRegionName());
put.add(HConstants.CATALOG_FAMILY, HConstants.REGIONINFO_QUALIFIER, Writables
.getBytes(newRegion.getRegionInfo()));
table.put(put);
HBaseAdmin admin = new HBaseAdmin(conf);
admin.getMaster().assign(newRegion.getRegionName(), true);
LOG.info("updated columns in row: " + Bytes.toString(newRegion.getRegionName()));
}
}
/** Instantiated to compact the meta region */
private static class OfflineMerger extends Merger {
private final List<HRegionInfo> metaRegions = new ArrayList<HRegionInfo>();
private final HRegion root;
OfflineMerger(Configuration conf, FileSystem fs) throws IOException {
super(conf, fs, HConstants.META_TABLE_NAME);
Path rootTableDir = HTableDescriptor.getTableDir(fs.makeQualified(new Path(conf
.get(HConstants.HBASE_DIR))), HConstants.ROOT_TABLE_NAME);
// Scan root region to find all the meta regions
root = HRegion.newHRegion(rootTableDir, hlog, fs, conf, HRegionInfo.ROOT_REGIONINFO,
null);
root.initialize();
Scan scan = new Scan();
scan.addColumn(HConstants.CATALOG_FAMILY, HConstants.REGIONINFO_QUALIFIER);
InternalScanner rootScanner = root.getScanner(scan);
try {
List<KeyValue> results = new ArrayList<KeyValue>();
while (rootScanner.next(results)) {
for (KeyValue kv : results) {
HRegionInfo info = Writables.getHRegionInfoOrNull(kv.getValue());
if (info != null) {
metaRegions.add(info);
}
}
}
} finally {
rootScanner.close();
try {
root.close();
} catch (IOException e) {
LOG.error(e);
}
}
}
@Override
protected HRegionInfo[] next() {
HRegionInfo[] results = null;
if (metaRegions.size() > 0) {
results = metaRegions.toArray(new HRegionInfo[metaRegions.size()]);
metaRegions.clear();
}
return results;
}
@Override
protected void updateMeta(final byte[] oldRegion1, final byte[] oldRegion2,
HRegion newRegion) throws IOException {
byte[][] regionsToDelete = { oldRegion1, oldRegion2 };
for (int r = 0; r < regionsToDelete.length; r++) {
Delete delete = new Delete(regionsToDelete[r]);
delete.deleteColumns(HConstants.CATALOG_FAMILY, HConstants.REGIONINFO_QUALIFIER);
delete.deleteColumns(HConstants.CATALOG_FAMILY, HConstants.SERVER_QUALIFIER);
delete.deleteColumns(HConstants.CATALOG_FAMILY, HConstants.STARTCODE_QUALIFIER);
delete.deleteColumns(HConstants.CATALOG_FAMILY, HConstants.SPLITA_QUALIFIER);
delete.deleteColumns(HConstants.CATALOG_FAMILY, HConstants.SPLITB_QUALIFIER);
root.delete(delete, null, true);
LOG.info("updated columns in row: " + Bytes.toString(regionsToDelete[r]));
}
HRegionInfo newInfo = newRegion.getRegionInfo();
newInfo.setOffline(true);
Put put = new Put(newRegion.getRegionName());
put.add(HConstants.CATALOG_FAMILY, HConstants.REGIONINFO_QUALIFIER, Writables
.getBytes(newInfo));
root.put(put);
LOG.info("updated columns in row: " + Bytes.toString(newRegion.getRegionName()));
}
}
}
发表评论
-
hbase关于bloom filter使用
2013-08-14 11:36 16551.Bloomfilter的原理? 可参考 http://hi ... -
hbase常用运维命令
2012-03-09 10:21 01.停止运行某个RegionServer,在该RegionSe ... -
HBase bulkload的一个bug定位
2011-12-30 11:13 2254在HBase bulkload过程中,如果rowkey和ver ... -
HBase 不常用命令列表
2011-12-30 10:38 11811.查看hfile的内容 hbase org.apache. ... -
Hbase的Region Compact算法实现分析
2011-12-08 15:06 4590Hbase的Region Compact算法属于一种多路归并的 ... -
Hbase region split源代码阅读笔记
2011-12-07 00:03 3159客户端 1. HbaseAdmin.split(final b ... -
Hbase bulkload源代码阅读笔记
2011-12-06 00:21 59991.LoadIncrementalHFiles.doBulkL ... -
Hbase HLog源代码阅读笔记
2011-11-03 18:01 2822HLog 当客户端往RegionServer上提交了一个更新操 ... -
Hbase WAL原理学习
2011-10-31 20:19 62241.概述 客户端往RegionServer端提交数据的时候,会 ... -
Hbase put过程源代码阅读笔记
2011-10-27 10:22 3411客户端 1.HTable.put for (Put p ... -
Hbase region compact源代码阅读笔记
2011-10-08 19:49 2894边缘代码略过,核心代 ...
相关推荐
介绍了大数据平台如何将hdfs中的分布式文件导入hbase 。源代码在cloudera-SCM 的cdh 4.8.1产品环境中验证通过。
《hbase权威指南》随书示例源代码.方便学习
hbase权威指南.源代码.绝对经典。。 下载后评分+评论,即可返回分数.
hbase0.94java源代码 希望对大家有帮助
VC代码 hbase1.0 (实用代码源).rarVC代码 hbase1.0 (实用代码源).rarVC代码 hbase1.0 (实用代码源).rarVC代码 hbase1.0 (实用代码源).rarVC代码 hbase1.0 (实用代码源).rarVC代码 hbase1.0 (实用代码源).rarVC代码 ...
基于Hadoop的大数据编程,主要涉及HDFS,MapRedue,HBase+源代码+文档说明 -------- 不懂运行,下载完可以私聊问,可远程教学 该资源内项目源码是个人的毕设,代码都测试ok,都是运行成功后才上传资源,答辩评审平均分...
hbase权威指南书中的源代码, hbase the definitive guide 源代码
Hbase权威指南 随书源代码 源码包 绝对完整版 maven工程,带pom文件,可以直接作为一个完整工程导入eclipse等ide。
hbase权威指南 源代码 英文
hbase源码,适合研究分析底层实现。对hbase的原理的理解很有好处
hbase 的java代码 集合 hbase 0.96
Hbase HLog源代码阅读笔记 HBase是一个分布式的、面向列的开源数据库,该技术来源于 Fay Chang 所撰写的Google论文“Bigtable:一个结构化数据的分布式存储系统”。就像Bigtable利用了Google文件系统(File System)...
hbase操作必备客户端源代码
java操作Hbase之Hbase专用过滤器PageFilter的使用源代码,附带全部所需源代码,欢迎下载学习。
- 不懂运行,下载完可以私聊问...3、如果基础还行,也可在此代码基础上进行修改,以实现其他功能,也可用于毕设、课设、作业等。 下载后请首先打开README.md文件(如有),仅供学习参考, 切勿用于商业用途。 --------
使用Java API连接虚拟机HBase并进行数据库操作,Java源代码
HBase基本操作 增删改查 java代码 要使用须导入对应的jar包
本代码是java链接hbase数据库并对hbase进行增删改查操作的实例代码,包括批量操作
NULL 博文链接:https://xaocaotanghui.iteye.com/blog/2154210
HBase研究 HBase数据库源代码学习研究(包括代码注释,文档,用于代码分析的测试用例)