1.3 被遗弃的hbase客户端使用代码
被遗弃的创建方式一:直接通过HTable(Configuration conf, final String tableName)创建
Configuration configuration = HBaseConfiguration.create();
configuration.set("hbase.zookeeper.property.clientPort", "2181");
configuration.set("hbase.client.write.buffer", "2097152");
configuration.set("hbase.zookeeper.quorum","192.168.199.31,192.168.199.32,192.168.199.33,192.168.199.34,192.168.199.35");
Table table = new HTable(configuration, "tableName");
//3177不是我杜撰的,是2*hbase.client.write.buffer/put.heapSize()计算出来的
int bestBathPutSize = 3177;
try {
// Use the table as needed, for a single operation and a single thread
// construct List<Put> putLists
List<Put> putLists = new ArrayList<Put>();
for(int count=0;count<100000;count++){
Put put = new Put(rowkey.getBytes());
put.addImmutable("columnFamily1".getBytes(), "columnName1".getBytes(), "columnValue1".getBytes());
put.addImmutable("columnFamily1".getBytes(), "columnName2".getBytes(), "columnValue2".getBytes());
put.addImmutable("columnFamily1".getBytes(), "columnName3".getBytes(), "columnValue3".getBytes());
put.setDurability(Durability.SKIP_WAL);
putLists.add(put);
if(putLists.size()==(bestBathPutSize-1)){
//达到最佳大小值了,马上提交一把
table.put(putLists);
putLists.clear();
}
}
//剩下的未提交数据,最后做一次提交
table.put(putLists)
} finally {
table.close();
connection.close();
}
被遗弃的方式二:通过HConnectionManager.createConnection(Configuration conf)获取HTableInterface
Configuration configuration = HBaseConfiguration.create();
configuration.set("hbase.zookeeper.property.clientPort", "2181");
configuration.set("hbase.client.write.buffer", "2097152");
configuration.set("hbase.zookeeper.quorum","192.168.199.31,192.168.199.32,192.168.199.33,192.168.199.34,192.168.199.35");
HConnection connection = HConnectionManager.createConnection(configuration);
HTableInterface table = connection.getTable(TableName.valueOf("tableName"));
//3177不是我杜撰的,是2*hbase.client.write.buffer/put.heapSize()计算出来的
int bestBathPutSize = 3177;
try {
// Use the table as needed, for a single operation and a single thread
// construct List<Put> putLists
List<Put> putLists = new ArrayList<Put>();
for(int count=0;count<100000;count++){
Put put = new Put(rowkey.getBytes());
put.addImmutable("columnFamily1".getBytes(), "columnName1".getBytes(), "columnValue1".getBytes());
put.addImmutable("columnFamily1".getBytes(), "columnName2".getBytes(), "columnValue2".getBytes());
put.addImmutable("columnFamily1".getBytes(), "columnName3".getBytes(), "columnValue3".getBytes());
put.setDurability(Durability.SKIP_WAL);
putLists.add(put);
if(putLists.size()==(bestBathPutSize-1)){
//达到最佳大小值了,马上提交一把
table.put(putLists);
putLists.clear();
}
}
//剩下的未提交数据,最后做一次提交
table.put(putLists)
} finally {
table.close();
connection.close();
}
2.hbase客户端源码解读
前面我们说过,推荐的使用hbase客户端的方式如下:
Connection connection = ConnectionFactory.createConnection(configuration);
Table table = connection.getTable(TableName.valueOf("tableName"));
那源代码的查看就从这两行代码开始,先来看下ConnectionFactory.createConnection(configuration)
2.1 ConnectionFactory.createConnection(Configuration conf)
先看下createConnection(Configuration conf)的源代码,如下:
public static Connection createConnection(Configuration conf) throws IOException {
return createConnection(conf, null, null);
}
传入我们构造的Configuration对象,然后调用了ConnectionFactory.createConnection(Configuration conf, ExecutorService pool, User user),继续看ConnectionFactory.createConnection(Configuration conf, ExecutorService pool, User user)的源代码,如下:
public static Connection createConnection(Configuration conf, ExecutorService pool, User user)
throws IOException {
//因为上面传入的user为null,这里代码不会执行
if (user == null) {
UserProvider provider = UserProvider.instantiate(conf);
user = provider.getCurrent();
}
return createConnection(conf, false, pool, user);
}
这里继续调用了ConnectionFactory.createConnection(final Configuration conf, final boolean managed, final ExecutorService pool, final User user),那么我们继续看下相关代码,如下:
static Connection createConnection(final Configuration conf, final boolean managed, final ExecutorService pool, final User user)
throws IOException {
//默认HBASE_CLIENT_CONNECTION_IMPL = "hbase.client.connection.impl"
//hbase.client.connection.impl供hbase使用者实现自己的hbase链接实现类并配置进来使用
//默认hbase已经提供了实现,无需实现,那么这里就取默认实现ConnectionManager.HConnectionImplementation.class.getName()
//默认hbase的connection实现类也即HConnectionImplementation类
String className = conf.get(HConnection.HBASE_CLIENT_CONNECTION_IMPL,ConnectionManager.HConnectionImplementation.class.getName());
Class<?> clazz = null;
try {
clazz = Class.forName(className);
} catch (ClassNotFoundException e) {
throw new IOException(e);
}
try {
// Default HCM#HCI is not accessible; make it so before invoking.
//这里调用HConnectionImplementation类的构造方法HConnectionImplementation(Configuration conf, boolean managed, ExecutorService pool, User user)
Constructor<?> constructor = clazz.getDeclaredConstructor(Configuration.class, boolean.class, ExecutorService.class, User.class);
constructor.setAccessible(true);
return (Connection) constructor.newInstance(conf, managed, pool, user);
} catch (Exception e) {
throw new IOException(e);
}
}
}
上面的代码默认调用ConnectionManager.HConnectionImplementation类返回Connection对象,继续跟踪HConnectionImplementation(Configuration conf, boolean managed, ExecutorService pool, User user)代码:
HConnectionImplementation(Configuration conf, boolean managed, ExecutorService pool, User user) throws IOException {
//这里代码我们需要重点关注
this(conf);
//这里this.user=null
this.user = user;
//这里this.batchPool=null
this.batchPool = pool;
//这里this.managed=false
this.managed = managed;
//这里setupRegistry()默认从hbase.client.registry.impl获取客户端使用者实现的zookeeper注册类,没有配置就默认创建ZooKeeperRegistry类对象并设置,这个类非常重要,客户端与zookeeper的交互类就由此类负责
this.registry = setupRegistry();
//默认通过ZooKeeperRegistry对象从zookeeper获取hbase集群的clusterId
retrieveClusterId();
//如果Configuration没配置hbase.rpc.client.impl就默认创建RpcClientImpl并设置给this.rpcClient
this.rpcClient = RpcClientFactory.createClient(this.conf, this.clusterId, this.metrics);
this.rpcControllerFactory = RpcControllerFactory.instantiate(conf);
// Do we publish the status?
//如果Configuration没配置hbase.status.published就默认设置shouldListen=false
boolean shouldListen = conf.getBoolean(HConstants.STATUS_PUBLISHED, HConstants.STATUS_PUBLISHED_DEFAULT);
//如果Configuration没配置hbase.status.listener.class就默认创建MulticastListener对象并设置给listenerClass
Class<? extends ClusterStatusListener.Listener> listenerClass = conf.getClass(ClusterStatusListener.STATUS_LISTENER_CLASS, ClusterStatusListener.DEFAULT_STATUS_LISTENER_CLASS, ClusterStatusListener.Listener.class);
if (shouldListen) {
if (listenerClass == null) {
LOG.warn(HConstants.STATUS_PUBLISHED + " is true, but " + ClusterStatusListener.STATUS_LISTENER_CLASS + " is not set - not listening status");
} else {
//这里通过hbase事件监听器监视hbase服务端事件,当hbase服务端服务不可用时,调用rpcClient.cancelConnections关闭链接
clusterStatusListener = new ClusterStatusListener(
new ClusterStatusListener.DeadServerHandler() {
@Override
public void newDead(ServerName sn) {
clearCaches(sn);
rpcClient.cancelConnections(sn);
}
}, conf, listenerClass);
}
}
}
上面的代码我们主要关注this(conf);另外一个需要注意的就是方法setupRegistry(),setupRegistry()这里默认设置的是org.apache.hadoop.hbase.client.ZooKeeperRegistry,这一行并将在后面继续分析,其它的代码都比较简单,我在上面代码中已经做代码注释,继续看this(conf)代码:
protected HConnectionImplementation(Configuration conf) {
//这里把客户端使用者传入的Configuration赋值给this.conf
this.conf = conf;
//这里HConnectionImplementation基于我们传入的Configuration构建了自己的Configuration类对象this.connectionConfig
this.connectionConfig = new ConnectionConfiguration(conf);
this.closed = false;
//客户端使用者的Configuration没有配置hbase.client.pause,那么就设置默认值this.pause=100
this.pause = conf.getLong(HConstants.HBASE_CLIENT_PAUSE, HConstants.DEFAULT_HBASE_CLIENT_PAUSE);
//客户端使用者的Configuration没有配置hbase.meta.replicas.use,那么就设置默认值this.useMetaReplicas=false
this.useMetaReplicas = conf.getBoolean(HConstants.USE_META_REPLICAS, HConstants.DEFAULT_USE_META_REPLICAS);
//从this.connectionConfig里获取值设置,而客户端使用者的Configuration没有配置hbase.client.retries.number就默认设置this.numTries=31
this.numTries = connectionConfig.getRetriesNumber();
//客户端使用者的Configuration没有配置hbase.rpc.timeout,那么就设置默认值this.rpcTimeout=60000毫秒
this.rpcTimeout = conf.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY, HConstants.DEFAULT_HBASE_RPC_TIMEOUT);
if (conf.getBoolean(CLIENT_NONCES_ENABLED_KEY, true)) {
synchronized (nonceGeneratorCreateLock) {
if (ConnectionManager.nonceGenerator == null) {
ConnectionManager.nonceGenerator = new PerClientRandomNonceGenerator();
}
this.nonceGenerator = ConnectionManager.nonceGenerator;
}
} else {
this.nonceGenerator = new NoNonceGenerator();
}
//跟踪region的统计信息
stats = ServerStatisticTracker.create(conf);
//hbase客户端异步操作类
this.asyncProcess = createAsyncProcess(this.conf);
this.interceptor = (new RetryingCallerInterceptorFactory(conf)).build();
this.rpcCallerFactory = RpcRetryingCallerFactory.instantiate(conf, interceptor, this.stats);
this.backoffPolicy = ClientBackoffPolicyFactory.create(conf);
if (conf.getBoolean(CLIENT_SIDE_METRICS_ENABLED_KEY, false)) {
this.metrics = new MetricsConnection(this);
} else {
this.metrics = null;
}
this.hostnamesCanChange = conf.getBoolean(RESOLVE_HOSTNAME_ON_FAIL_KEY, true);
this.metaCache = new MetaCache(this.metrics);
}
上面代码比较重要的一点是,尽管客户端传入了Configuration,但是HConnectionImplementation不会直接使用客户端传入的Configuration,而是基于客户端传入的Configuration构建了自己的Configuration对象,原因是客户端传入的Configuration对象只给了部分值,很多其它值都未给出,那么HConnectionImplementation就有必要创建自己的Configuration,首先构建自己默认的Configuration,然后把客户端已经设置的Configuration的相关值覆盖那些默认值,客户端没设置的值就使用默认值,我们继续看下this.connectionConfig = new ConnectionConfiguration(conf)的源代码:
ConnectionConfiguration(Configuration conf) {
//客户端的Configuration没有配置hbase.client.pause,那么就设置默认值this.writeBufferSize=2097152
this.writeBufferSize = conf.getLong(WRITE_BUFFER_SIZE_KEY, WRITE_BUFFER_SIZE_DEFAULT);
//客户端的Configuration没有配置hbase.client.write.buffer,那么就设置默认值this.metaOperationTimeout=1200000
this.metaOperationTimeout = conf.getInt(HConstants.HBASE_CLIENT_META_OPERATION_TIMEOUT, HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT);
//客户端的Configuration没有配置hbase.client.meta.operation.timeout,那么就设置默认值this.operationTimeout=1200000
this.operationTimeout = conf.getInt(HConstants.HBASE_CLIENT_OPERATION_TIMEOUT, HConstants.DEFAULT_HBASE_CLIENT_OPERATION_TIMEOUT);
//客户端的Configuration没有配置hbase.client.operation.timeout,那么就设置默认值this.scannerCaching=Integer.MAX_VALUE
this.scannerCaching = conf.getInt(HConstants.HBASE_CLIENT_SCANNER_CACHING, HConstants.DEFAULT_HBASE_CLIENT_SCANNER_CACHING);
//客户端的Configuration没有配置hbase.client.scanner.max.result.size,那么就设置默认值this.scannerMaxResultSize=2 * 1024 * 1024
this.scannerMaxResultSize = conf.getLong(HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY, HConstants.DEFAULT_HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE);
//客户端的Configuration没有配置hbase.client.primaryCallTimeout.get,那么就设置默认值this.primaryCallTimeoutMicroSecond=10000
this.primaryCallTimeoutMicroSecond = conf.getInt("hbase.client.primaryCallTimeout.get", 10000); // 10000ms
//客户端的Configuration没有配置hbase.client.replicaCallTimeout.scan,那么就设置默认值this.replicaCallTimeoutMicroSecondScan=1000000
this.replicaCallTimeoutMicroSecondScan = conf.getInt("hbase.client.replicaCallTimeout.scan", 1000000); // 1000000ms
//客户端的Configuration没有配置hbase.client.retries.number,那么就设置默认值this.retries=31
this.retries = conf.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
//客户端的Configuration没有配置hbase.client.keyvalue.maxsize,那么就设置默认值this.maxKeyValueSize=-1
this.maxKeyValueSize = conf.getInt(MAX_KEYVALUE_SIZE_KEY, MAX_KEYVALUE_SIZE_DEFAULT);
}
上面的代码主要是初始化HConnectionImplementation自己的Configuration类型属性this.connectionConfig,默认客户端不设置属性值,这里创建的this.connectionConfig就使用默认值,这里将hbase客户端默认值抽取如下:
- hbase.client.write.buffer 默认2097152Byte,也即2MB
- hbase.client.meta.operation.timeout 默认1200000毫秒
- hbase.client.operation.timeout 默认1200000毫秒
- hbase.client.scanner.caching 默认Integer.MAX_VALUE
- hbase.client.scanner.max.result.size 默认2MB
- hbase.client.primaryCallTimeout.get 默认10000毫秒
- hbase.client.replicaCallTimeout.scan 默认1000000毫秒
- hbase.client.retries.number 默认31次
- hbase.client.keyvalue.maxsize 默认-1,不限制
- hbase.client.ipc.pool.type
- hbase.client.ipc.pool.size
- hbase.client.pause 100
- hbase.client.max.total.tasks 100
- hbase.client.max.perserver.tasks 2
- hbase.client.max.perregion.tasks 1
- hbase.client.instance.id
- hbase.client.scanner.timeout.period 60000
- hbase.client.rpc.codec
- hbase.regionserver.lease.period 被hbase.client.scanner.timeout.period代替,60000
- hbase.client.fast.fail.mode.enabled FALSE
- hbase.client.fastfail.threshold 60000
- hbase.client.fast.fail.cleanup.duration 600000
- hbase.client.fast.fail.interceptor.impl
- hbase.client.backpressure.enabled false
2.2 与zookeeper交互的ZooKeeperRegistry
上面我们分析知道客户端使用者传入的Configuration只有设置的值才会在客户端上生效,而未设置的值则交由默认值设置,另外一个非常重要的就是刚才所提到的与zookeeper交互的类org.apache.hadoop.hbase.client.ZooKeeperRegistry
package org.apache.hadoop.hbase.client;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.RegionLocations;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.zookeeper.MetaTableLocator;
import org.apache.hadoop.hbase.zookeeper.ZKClusterId;
import org.apache.hadoop.hbase.zookeeper.ZKTableStateClientSideReader;
import org.apache.hadoop.hbase.zookeeper.ZKUtil;
import org.apache.zookeeper.KeeperException;
/**
* A cluster registry that stores to zookeeper.
*/
class ZooKeeperRegistry implements Registry {
private static final Log LOG = LogFactory.getLog(ZooKeeperRegistry.class);
// hbase连接,在初始化函数中会进行设置
ConnectionManager.HConnectionImplementation hci;
@Override
public void init(Connection connection) {
if (!(connection instanceof ConnectionManager.HConnectionImplementation)) {
throw new RuntimeException("This registry depends on HConnectionImplementation");
}
//设置hbase连接
this.hci = (ConnectionManager.HConnectionImplementation)connection;
}
@Override
public RegionLocations getMetaRegionLocation() throws IOException {
//通过hbase连接中的Configuration获取zookeeper地址后,通过hbase连接获取与zookeeper交互的ZooKeeperKeepAliveConnection
ZooKeeperKeepAliveConnection zkw = hci.getKeepAliveZooKeeperWatcher();
try {
if (LOG.isTraceEnabled()) {
LOG.trace("Looking up meta region location in ZK," + " connection=" + this);
}
//从zookeeper中获取所有的hbase region元数据信息
List<ServerName> servers = new MetaTableLocator().blockUntilAvailable(zkw, hci.rpcTimeout, hci.getConfiguration());
if (LOG.isTraceEnabled()) {
if (servers == null) {
LOG.trace("Looked up meta region location, connection=" + this + "; servers = null");
} else {
StringBuilder str = new StringBuilder();
for (ServerName s : servers) {
str.append(s.toString());
str.append(" ");
}
LOG.trace("Looked up meta region location, connection=" + this + "; servers = " + str.toString());
}
}
if (servers == null) return null;
//组装hbase RegionLocations数组进行返回
HRegionLocation[] locs = new HRegionLocation[servers.size()];
int i = 0;
for (ServerName server : servers) {
HRegionInfo h = RegionReplicaUtil.getRegionInfoForReplica(HRegionInfo.FIRST_META_REGIONINFO, i);
if (server == null) locs[i++] = null;
else locs[i++] = new HRegionLocation(h, server, 0);
}
return new RegionLocations(locs);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return null;
} finally {
zkw.close();
}
}
private String clusterId = null;
@Override
public String getClusterId() {
if (this.clusterId != null) return this.clusterId;
// No synchronized here, worse case we will retrieve it twice, that's
// not an issue.
ZooKeeperKeepAliveConnection zkw = null;
try {
zkw = hci.getKeepAliveZooKeeperWatcher();
this.clusterId = ZKClusterId.readClusterIdZNode(zkw);
if (this.clusterId == null) {
LOG.info("ClusterId read in ZooKeeper is null");
}
} catch (KeeperException e) {
LOG.warn("Can't retrieve clusterId from Zookeeper", e);
} catch (IOException e) {
LOG.warn("Can't retrieve clusterId from Zookeeper", e);
} finally {
if (zkw != null) zkw.close();
}
return this.clusterId;
}
@Override
public boolean isTableOnlineState(TableName tableName, boolean enabled)
throws IOException {
ZooKeeperKeepAliveConnection zkw = hci.getKeepAliveZooKeeperWatcher();
try {
if (enabled) {
return ZKTableStateClientSideReader.isEnabledTable(zkw, tableName);
}
return ZKTableStateClientSideReader.isDisabledTable(zkw, tableName);
} catch (KeeperException e) {
throw new IOException("Enable/Disable failed", e);
} catch (InterruptedException e) {
throw new InterruptedIOException();
} finally {
zkw.close();
}
}
@Override
public int getCurrentNrHRS() throws IOException {
ZooKeeperKeepAliveConnection zkw = hci.getKeepAliveZooKeeperWatcher();
try {
// We go to zk rather than to master to get count of regions to avoid
// HTable having a Master dependency. See HBase-2828
return ZKUtil.getNumberOfChildren(zkw, zkw.rsZNode);
} catch (KeeperException ke) {
throw new IOException("Unexpected ZooKeeper exception", ke);
} finally {
zkw.close();
}
}
}
这个类非常重要,因为所有的与zookeeper的交互都由它来完成。
2.3 HConnectionImplementation.getTable(TableName tableName)
前面我们说过,推荐的使用hbase客户端的方式如下:
Connection connection = ConnectionFactory.createConnection(configuration);
Table table = connection.getTable(TableName.valueOf("tableName"));
上面2.1中已经知悉默认connection实现是HConnectionImplementation,那么这里我们继续跟踪HConnectionImplementation.getTable(TableName tableName)方法,代码如下:
public HTableInterface getTable(TableName tableName) throws IOException {
return getTable(tableName, getBatchPool());
}
继续看HConnectionImplementation.getTable(TableName tableName, ExecutorService pool)的代码:
public HTableInterface getTable(TableName tableName, ExecutorService pool) throws IOException {
//默认managed=false
if (managed) {
throw new NeedUnmanagedConnectionException();
}
return new HTable(tableName, this, connectionConfig, rpcCallerFactory, rpcControllerFactory, pool);
}
继续看HTable的构造方法HTable(TableName tableName, final ClusterConnection connection, final ConnectionConfiguration tableConfig, final RpcRetryingCallerFactory rpcCallerFactory, final RpcControllerFactory rpcControllerFactory, final ExecutorService pool),代码如下:
public HTable(TableName tableName, final ClusterConnection connection, final ConnectionConfiguration tableConfig, final RpcRetryingCallerFactory rpcCallerFactory, final RpcControllerFactory rpcControllerFactory, final ExecutorService pool) throws IOException {
if (connection == null || connection.isClosed()) {
throw new IllegalArgumentException("Connection is null or closed.");
}
//设置hbase数据表名
this.tableName = tableName;
//调用close方法时,默认不关闭连接,这一点非常重要,默认调用table.close()是不会关闭之前创建的connection的,这一点在后面的table.close()里会介绍
this.cleanupConnectionOnClose = false;
//设置this.connection值为HConnectionImplementation创建的connection实现类
this.connection = connection;
//从HConnectionImplementation获取客户端传入的configuration对象
this.configuration = connection.getConfiguration();
//从HConnectionImplementation获取HConnectionImplementation基于客户端传入的configuration创建的configuration对象
this.connConfiguration = tableConfig;
//从HConnectionImplementation获取pool,HConnectionImplementation的默认pool为this.batchPool = getThreadPool(conf.getInt("hbase.hconnection.threads.max", 256)
this.pool = pool;
if (pool == null) {
this.pool = getDefaultExecutor(this.configuration);
this.cleanupPoolOnClose = true;
} else {
//在HConnectionImplementation中已经初始化了this.batchPool = getThreadPool(conf.getInt("hbase.hconnection.threads.max", 256),所以这里会设置cleanupPoolOnClose,默认也不会关闭线程池
this.cleanupPoolOnClose = false;
}
this.rpcCallerFactory = rpcCallerFactory;
this.rpcControllerFactory = rpcControllerFactory;
//这个方法我们后面重点关注,其根据客户端传入的Configuration初始化HTable的参数
this.finishSetup();
}
上面的代码我已经加了注释,需要注意的是cleanupConnectionOnClose属性,该属性默认值为false,在调用table.close()方法时候,只是关闭了table而已但table后面的connection是没有关闭的,再者是属性cleanupPoolOnClose,虽然我们没有传入线程池,但是HConnectionImplementation会自己创建线程池this.batchPool = getThreadPool(conf.getInt("hbase.hconnection.threads.max", 256)传过来使用,所以这里会设置this.cleanupPoolOnClose = false,默认在table.close()调用时候,也不会关闭线程池,那么这里这里继续跟踪上面代码最后的this.finishSetup(),代码如下:
private void finishSetup() throws IOException {
//HTable的属性connConfiguration若为空,就基于客户端传入的Configuration构建新的connConfiguration
if (connConfiguration == null) {
connConfiguration = new ConnectionConfiguration(configuration);
}
//HTable的属性设置
this.operationTimeout = tableName.isSystemTable() ? connConfiguration.getMetaOperationTimeout() : connConfiguration.getOperationTimeout();
this.scannerCaching = connConfiguration.getScannerCaching();
this.scannerMaxResultSize = connConfiguration.getScannerMaxResultSize();
if (this.rpcCallerFactory == null) {
this.rpcCallerFactory = connection.getNewRpcRetryingCallerFactory(configuration);
}
if (this.rpcControllerFactory == null) {
this.rpcControllerFactory = RpcControllerFactory.instantiate(configuration);
}
// puts need to track errors globally due to how the APIs currently work.
//hbase的异步操作类
multiAp = this.connection.getAsyncProcess();
this.closed = false;
//hbase的region操作工具类
this.locator = new HRegionLocator(tableName, connection);
}
经过上面的分析,我们有必要看下table.close()的源代码:
public void close() throws IOException {
//如果已经关闭了,直接返回
if (this.closed) {
return;
}
//关闭前做最后一次提交
flushCommits();
//默认在构造HTable时候,cleanupPoolOnClose=false,这里不会去关闭线程池
if (cleanupPoolOnClose) {
this.pool.shutdown();
try {
boolean terminated = false;
do {
// wait until the pool has terminated
terminated = this.pool.awaitTermination(60, TimeUnit.SECONDS);
} while (!terminated);
} catch (InterruptedException e) {
this.pool.shutdownNow();
LOG.warn("waitForTermination interrupted");
}
}
//默认在构造HTable时候,cleanupConnectionOnClose=false,这里不会去关闭table持有的connection
if (cleanupConnectionOnClose) {
if (this.connection != null) {
this.connection.close();
}
}
this.closed = true;
}
2.4 HTable.put(final List<Put> puts)
我们已经通过如下代码:
Connection connection = ConnectionFactory.createConnection(configuration);
Table table = connection.getTable(TableName.valueOf("tableName"));
创建了connection,其默认实现类为org.apache.hadoop.hbase.client.ConnectionManager.HConnectionImplementation,然后创建了table,其默认实现类为org.apache.hadoop.hbase.client.HTable,那么接下来就是分析客户端的批量提交方法:HTable.put(final List<Put> puts),代码如下:
public void put(final List<Put> puts) throws IOException {
//根据设置的缓存大小,达到缓存相关值就进行批量提交
getBufferedMutator().mutate(puts);
//不管有无数据未提交,默认autoFlush=true,那么就最后提交一次
if (autoFlush) {
flushCommits();
}
}
这里先看下HTable.getBufferedMutator()源代码:
BufferedMutator getBufferedMutator() throws IOException {
if (mutator == null) {
//从HConnectionImplementation获取pool,HConnectionImplementation的默认pool为this.batchPool = getThreadPool(conf.getInt("hbase.hconnection.threads.max", 256)
//根据hbase.client.write.buffer设置的值,默认2MB,构造缓冲区
this.mutator = (BufferedMutatorImpl) connection.getBufferedMutator(
new BufferedMutatorParams(tableName)
.pool(pool)
.writeBufferSize(connConfiguration.getWriteBufferSize())
.maxKeyValueSize(connConfiguration.getMaxKeyValueSize())
);
}
return mutator;
}
上面的代码默认构造了一个BufferedMutatorImpl类并返回,继续跟踪BufferedMutatorImpl的方法mutate(List<? extends Mutation> ms)
public void mutate(List<? extends Mutation> ms) throws InterruptedIOException, RetriesExhaustedWithDetailsException {
//如果BufferedMutatorImpl已经关闭,直接退出返回
if (closed) {
throw new IllegalStateException("Cannot put when the BufferedMutator is closed.");
}
//这里先不断循环累计提交的List<Put>记录所占的空间,放置到toAddSize
long toAddSize = 0;
for (Mutation m : ms) {
if (m instanceof Put) {
validatePut((Put) m);
}
toAddSize += m.heapSize();
}
// This behavior is highly non-intuitive... it does not protect us against
// 94-incompatible behavior, which is a timing issue because hasError, the below code
// and setter of hasError are not synchronized. Perhaps it should be removed.
if (ap.hasError()) {
//设置BufferedMutatorImpl当前记录的提交记录所占空间值为toAddSize
currentWriteBufferSize.addAndGet(toAddSize);
//把提交的记录List<Put>放置到缓存对象writeAsyncBuffer,在为提交完成前先不进行清理
writeAsyncBuffer.addAll(ms);
//这里当捕获到异常时候,再进行异常前的一次数据提交
backgroundFlushCommits(true);
} else {
//设置BufferedMutatorImpl当前记录的提交记录所占空间值为toAddSize
currentWriteBufferSize.addAndGet(toAddSize);
//把提交的记录List<Put>放置到缓存对象writeAsyncBuffer,在为提交完成前先不进行清理
writeAsyncBuffer.addAll(ms);
}
// Now try and queue what needs to be queued.
// 如果当前提交的List<Put>记录所占空间大于hbase.client.write.buffer设置的值,默认2MB,那么就马上调用backgroundFlushCommits方法
// 如果小于hbase.client.write.buffer设置的值,那么就直接退出,啥也不做
while (currentWriteBufferSize.get() > writeBufferSize) {
backgroundFlushCommits(false);
}
}
上面的代码不断循环累计提交的List<Put>记录所占的空间,如果所占空间大于hbase.client.write.buffer设置的值,那么就马上调用backgroundFlushCommits(false)方法,否则啥也不做,如果出错就马上调用一次backgroundFlushCommits(true),所以我们很有必要继续跟踪BufferedMutatorImpl.backgroundFlushCommits(boolean synchronous)代码:
private void backgroundFlushCommits(boolean synchronous) throws InterruptedIOException, RetriesExhaustedWithDetailsException {
LinkedList<Mutation> buffer = new LinkedList<>();
// Keep track of the size so that this thread doesn't spin forever
long dequeuedSize = 0;
try {
//分析所有提交的List<Put>,Put是Mutation的实现
Mutation m;
//如果(hbase.client.write.buffer <= 0 || 0 < (whbase.client.write.buffer * 2) || synchronous)&& writeAsyncBuffer里仍然有Mutation对象
//那么就不断计算所占空间大小dequeuedSize
//currentWriteBufferSize的大小则递减
while ((writeBufferSize <= 0 || dequeuedSize < (writeBufferSize * 2) || synchronous) && (m = writeAsyncBuffer.poll()) != null) {
buffer.add(m);
long size = m.heapSize();
dequeuedSize += size;
currentWriteBufferSize.addAndGet(-size);
}
//backgroundFlushCommits(false)时候,当List<Put>,这里不会进入
if (!synchronous && dequeuedSize == 0) {
return;
}
//backgroundFlushCommits(false)时候,这里会进入,并且不会等待结果返回
if (!synchronous) {
//不会等待结果返回
ap.submit(tableName, buffer, true, null, false);
if (ap.hasError()) {
LOG.debug(tableName + ": One or more of the operations have failed -"
+ " waiting for all operation in progress to finish (successfully or not)");
}
}
//backgroundFlushCommits(true)时候,这里会进入,并且会等待结果返回
if (synchronous || ap.hasError()) {
while (!buffer.isEmpty()) {
ap.submit(tableName, buffer, true, null, false);
}
//会等待结果返回
RetriesExhaustedWithDetailsException error = ap.waitForAllPreviousOpsAndReset(null);
if (error != null) {
if (listener == null) {
throw error;
} else {
this.listener.onException(error, this);
}
}
}
} finally {
//如果还有数据,那么给到外面最后提交
for (Mutation mut : buffer) {
long size = mut.heapSize();
currentWriteBufferSize.addAndGet(size);
dequeuedSize -= size;
writeAsyncBuffer.add(mut);
}
}
}
这里会调用ap.submit(tableName, buffer, true, null, false)直接提交,并且不会等待返回结果,而ap.submit(tableName, buffer, true, null, false)会调用AsyncProcess.submit(ExecutorService pool, TableName tableName,List<? extends Row> rows, boolean atLeastOne, Batch.Callback<CResult> callback,boolean needResults),这里源代码如下:
public <CResult> AsyncRequestFuture submit(TableName tableName, List<? extends Row> rows,
boolean atLeastOne, Batch.Callback<CResult> callback, boolean needResults)
throws InterruptedIOException {
return submit(null, tableName, rows, atLeastOne, callback, needResults);
}
public <CResult> AsyncRequestFuture submit(ExecutorService pool, TableName tableName, List<? extends Row> rows, boolean atLeastOne, Batch.Callback<CResult> callback, boolean needResults) throws InterruptedIOException {
//如果提交的记录数为0,就直接返回NO_REQS_RESULT
if (rows.isEmpty()) {
return NO_REQS_RESULT;
}
Map<ServerName, MultiAction<Row>> actionsByServer = new HashMap<ServerName, MultiAction<Row>>();
//依据提交的List<Put>的记录数构建retainedActions
List<Action<Row>> retainedActions = new ArrayList<Action<Row>>(rows.size());
NonceGenerator ng = this.connection.getNonceGenerator();
long nonceGroup = ng.getNonceGroup(); // Currently, nonce group is per entire client.
// Location errors that happen before we decide what requests to take.
List<Exception> locationErrors = null;
List<Integer> locationErrorRows = null;
//只要retainedActions不为空,那么就一直执行
do {
// Wait until there is at least one slot for a new task.
// 默认maxTotalConcurrentTasks=100,即最多100个异步线程用于处理元数据获取任务,如果超过100,就等待
waitForMaximumCurrentTasks(maxTotalConcurrentTasks - 1);
// Remember the previous decisions about regions or region servers we put in the
// final multi.
// 记录本次提交的List<Put>对应的region和regionserver
Map<HRegionInfo, Boolean> regionIncluded = new HashMap<HRegionInfo, Boolean>();
Map<ServerName, Boolean> serverIncluded = new HashMap<ServerName, Boolean>();
int posInList = -1;
Iterator<? extends Row> it = rows.iterator();
while (it.hasNext()) {
//这里默认传入一个Put对象,因为Put是Row的继承类
Row r = it.next();
//建立变量loc用来存储Put对象对应的region对应的元数据信息
HRegionLocation loc;
try {
if (r == null) {
throw new IllegalArgumentException("#" + id + ", row cannot be null");
}
// Make sure we get 0-s replica.
//取得Put对象对应的region元数据信息的所有备份信息,第一次调用时候会缓存中是没有元数据信息的,那么就会去链接zookeeper上查找,找到后就加入到缓存,下一次直接从缓存中获取
RegionLocations locs = connection.locateRegion(
tableName, r.getRow(), true, true, RegionReplicaUtil.DEFAULT_REPLICA_ID);
if (locs == null || locs.isEmpty() || locs.getDefaultRegionLocation() == null) {
throw new IOException("#" + id + ", no location found, aborting submit for"
+ " tableName=" + tableName + " rowkey=" + Bytes.toStringBinary(r.getRow()));
}
//取得Put对象对应的region元数据信息的所有备份信息数组中的第一个
loc = locs.getDefaultRegionLocation();
} catch (IOException ex) {
locationErrors = new ArrayList<Exception>();
locationErrorRows = new ArrayList<Integer>();
LOG.error("Failed to get region location ", ex);
// This action failed before creating ars. Retain it, but do not add to submit list.
// We will then add it to ars in an already-failed state.
retainedActions.add(new Action<Row>(r, ++posInList));
locationErrors.add(ex);
locationErrorRows.add(posInList);
it.remove();
break; // Backward compat: we stop considering actions on location error.
}
//这里判断是否可以操作,因为最多也就100个异步线程获取元数据信息,如果都忙就等待
if (canTakeOperation(loc, regionIncluded, serverIncluded)) {
Action<Row> action = new Action<Row>(r, ++posInList);
setNonce(ng, r, action);//
retainedActions.add(action);
// TODO: replica-get is not supported on this path
byte[] regionName = loc.getRegionInfo().getRegionName();
//把同一个区的提交任务进行收集,这里先只获知元数据信息,用于知道数据需要提交到哪个region和regionserver,最后循环外再做提交
addAction(loc.getServerName(), regionName, action, actionsByServer, nonceGroup);
it.remove();
}
}
} while (retainedActions.isEmpty() && atLeastOne && (locationErrors == null));
if (retainedActions.isEmpty()) return NO_REQS_RESULT;
// 这里已经知道数据该提交到哪个region和regionserver,就进行批量提交
return submitMultiActions(tableName, retainedActions, nonceGroup, callback, null, needResults, locationErrors, locationErrorRows, actionsByServer, pool);
}
上面代码会去寻找提交的List<Put>的每个Put对象对应的region是哪个,对应的regionserver是哪个,然后进行批量提交,这里要提到另外一个值hbase.client.max.total.tasks(默认值100,意思为客户端最大处理线程数),如果去请求Put对象对应的region是哪个和对应的regionserver是哪个的操作大于100,那么就要等待,我们回到最初的客户端批量提交代码:
public void put(final List<Put> puts) throws IOException {
//根据设置的缓存大小,达到缓存相关值就进行批量提交
getBufferedMutator().mutate(puts);
//不管有无数据未提交,默认autoFlush=true,那么就最后提交一次
if (autoFlush) {
flushCommits();
}
}
上面的分析可知,如果客户端提交的List<Put>所占空间满足不同条件会进行不同处理,总结如下:
- List<Put>所占空间<hbase.client.write.buffer:getBufferedMutator().mutate(puts)会直接退出,直接执行flushCommits()
- hbase.client.write.buffer<List<Put>所占空间<2*hbase.client.write.buffer:getBufferedMutator().mutate(puts)里面会执行backgroundFlushCommits(false),处理完后执行flushCommits()
- 2*hbase.client.write.buffer<List<Put>所占空间:getBufferedMutator().mutate(puts)里面会执行backgroundFlushCommits(false),多余的未提交数据会保留,然后执行flushCommits()
紧接着,如果HTable的属性autoFlush(默认为true),那么不管剩下的数据多少,也会进行最后一次提交数据到hbase服务端,这时候flushCommits()里调用的是getBufferedMutator().flush(),而getBufferedMutator().flush()调用的是BufferedMutatorImpl.backgroundFlushCommits(true),最后调用上面的ap.submit(tableName, buffer, true, null, false)并且会调用ap.waitForAllPreviousOpsAndReset(null)等待返回结果,至此hbase客户端批量提交的源代码分析完毕。
2.5.HConnectionImplementation.locateRegionInMeta
上面的代码HTable.put(final List<Put> puts)分析中我们需要关注另一个重要的信息,就是org.apache.hadoop.hbase.client.AsyncProcess的方法public <CResult> AsyncRequestFuture submit(TableName tableName, List<? extends Row> rows, boolean atLeastOne, Batch.Callback<CResult> callback, boolean needResults),在这个方法里有这么一段代码:
// 获取我们的数据表的region信息
RegionLocations locs = connection.locateRegion(tableName,r.getRow(), true, true, RegionReplicaUtil.DEFAULT_REPLICA_ID);
实质是调用了org.apache.hadoop.hbase.client.ConnectionManager.HConnectionImplementation的方法public RegionLocations locateRegion(final TableName tableName, final byte [] row, boolean useCache, boolean retry, int replicaId),这个方法加载了我们的hbase数据表的region信息,代码解释如下:
public RegionLocations locateRegion(final TableName tableName, final byte [] row, boolean useCache, boolean retry, int replicaId) throws IOException {
//如果当前连接已经关闭,抛出异常
if (this.closed) throw new IOException(toString() + " closed");
//如果客户端传入hbase数据表为空,抛出异常
if (tableName== null || tableName.getName().length == 0) {
throw new IllegalArgumentException("table name cannot be null or zero length");
}
//TableName.META_TABLE_NAME=hbase:meta(冒号前hbase为包名,meta为表名)
//我们传入的是我们自己的hbase数据表名,而不是hbase:meta,所以这里不会进入
if (tableName.equals(TableName.META_TABLE_NAME)) {
return locateMeta(tableName, useCache, replicaId);
} else {
// 这里的代码会进入
// 这里会去hbase的元数据信息表hbase:meta里去按照我们所给的数据表名和rowkey寻找我们的hbase数据表的region信息
return locateRegionInMeta(tableName, row, useCache, retry, replicaId);
}
}
我们继续关注locateRegionInMeta(tableName, row, useCache, retry, replicaId),代码注释如下:
/*
* 这里会去hbase的元数据信息表hbase:meta里去按照我们所给的数据表名和rowkey寻找我们的hbase数据表的region信息
*/
private RegionLocations locateRegionInMeta(TableName tableName, byte[] row, boolean useCache, boolean retry, int replicaId) throws IOException {
// 这里传入的useCache=true,所以会进入
if (useCache) {
//虽然进入了,但是第一次从缓存中找不到我们的数据表的相关信息
RegionLocations locations = getCachedLocation(tableName, row);
if (locations != null && locations.getRegionLocation(replicaId) != null) {
return locations;
}
}
//这里去元数据表hbase:meta中找数据,所以需要构造rowkey
// rowkey=tableName+我们传入的rowkey+"99999999999999"+前面字符的md5HashBytes
byte[] metaKey = HRegionInfo.createRegionName(tableName, row, HConstants.NINES, false);
//这里构造元数据表hbase:meta的查询scan
Scan s = new Scan();
s.setReversed(true);
s.setStartRow(metaKey);
s.setSmall(true);
s.setCaching(1);
if (this.useMetaReplicas) {
s.setConsistency(Consistency.TIMELINE);
}
//默认numTries=31次,无法从元数据表hbase:meta获取信息,那么就一直尝试31次
int localNumRetries = (retry ? numTries : 1);
for (int tries = 0; true; tries++) {
if (tries >= localNumRetries) {
throw new NoServerForRegionException("Unable to find region for " + Bytes.toStringBinary(row) + " in " + tableName + " after " + localNumRetries + " tries.");
}
if (useCache) {//这里虽然进入了,因为useCache=true,但是我们第一次还是无法从缓存拿到数据
RegionLocations locations = getCachedLocation(tableName, row);
if (locations != null && locations.getRegionLocation(replicaId) != null) {
return locations;
}
} else {
// If we are not supposed to be using the cache, delete any existing cached location
// so it won't interfere.
metaCache.clearCache(tableName, row);
}
// 因为缓存拿不到,那么就从元数据表hbase:meta获取region信息
try {
Result regionInfoRow = null;
ReversedClientScanner rcs = null;
try {
//这里很重要,告诉刚才构造的scan用于表TableName.META_TABLE_NAME,而TableName.META_TABLE_NAME=hbase:meta
rcs = new ClientSmallReversedScanner(conf, s, TableName.META_TABLE_NAME, this, rpcCallerFactory, rpcControllerFactory, getMetaLookupPool(), 0);
//好了,这里拿到了我们的数据表的regionInfoRow信息,regionInfoRow是元数据表hbase:meta中的一行数据
regionInfoRow = rcs.next();
} finally {
if (rcs != null) {
rcs.close();
}
}
if (regionInfoRow == null) {
throw new TableNotFoundException(tableName);
}
// 转换数据表的regionInfoRow信息为我们需要的HRegionLocation
RegionLocations locations = MetaTableAccessor.getRegionLocations(regionInfoRow);
if (locations == null || locations.getRegionLocation(replicaId) == null) {
throw new IOException("HRegionInfo was null in " + tableName + ", row=" + regionInfoRow);
}
//我们拿到了我们的hbase数据表的HRegionLocation,但是此时再做个检查,避免此时hbase宕机了或者已经split了或者拿错了
HRegionInfo regionInfo = locations.getRegionLocation(replicaId).getRegionInfo();
if (regionInfo == null) {
throw new IOException("HRegionInfo was null or empty in " + TableName.META_TABLE_NAME + ", row=" + regionInfoRow);
}
if (!regionInfo.getTable().equals(tableName)) {
throw new TableNotFoundException( "Table '" + tableName + "' was not found, got: " + regionInfo.getTable() + ".");
}
if (regionInfo.isSplit()) {
throw new RegionOfflineException("the only available region for" + " the required row is a split parent," + " the daughters should be online soon: " + regionInfo.getRegionNameAsString());
}
if (regionInfo.isOffline()) {
throw new RegionOfflineException("the region is offline, could" + " be caused by a disable table call: " + regionInfo.getRegionNameAsString());
}
ServerName serverName = locations.getRegionLocation(replicaId).getServerName();
if (serverName == null) {
throw new NoServerForRegionException("No server address listed " + "in " + TableName.META_TABLE_NAME + " for region " + regionInfo.getRegionNameAsString() + " containing row " + Bytes.toStringBinary(row));
}
if (isDeadServer(serverName)){
throw new RegionServerStoppedException("hbase:meta says the region "+ regionInfo.getRegionNameAsString()+" is managed by the server " + serverName + ", but it is dead.");
}
// 好了检查无误了,那么为了让下一次不要这么麻烦,先缓存起来,这样拿的也快
cacheLocation(tableName, locations);
// 好了,该返回region信息了
return locations;
} catch (TableNotFoundException e) {
// if we got this error, probably means the table just plain doesn't
// exist. rethrow the error immediately. this should always be coming
// from the HTable constructor.
throw e;
} catch (IOException e) {
ExceptionUtil.rethrowIfInterrupt(e);
if (e instanceof RemoteException) {
e = ((RemoteException)e).unwrapRemoteException();
}
if (tries < localNumRetries - 1) {
if (LOG.isDebugEnabled()) {
LOG.debug("locateRegionInMeta parentTable=" + TableName.META_TABLE_NAME + ", metaLocation=" + ", attempt=" + tries + " of " + localNumRetries + " failed; retrying after sleep of " + ConnectionUtils.getPauseTime(this.pause, tries) + " because: " + e.getMessage());
}
} else {
throw e;
}
// Only relocate the parent region if necessary
if(!(e instanceof RegionOfflineException || e instanceof NoServerForRegionException)) {
relocateRegion(TableName.META_TABLE_NAME, metaKey, replicaId);
}
}
//没找到,那么沉睡一段时间然后重试次数未到31次,那么继续循环找吧,直到找到,如果次数大于31,那么只有抛出异常
try{
Thread.sleep(ConnectionUtils.getPauseTime(this.pause, tries));
} catch (InterruptedException e) {
throw new InterruptedIOException("Giving up trying to location region in " + "meta: thread is interrupted.");
}
}
}
上述代码我们可以得知在首次org.apache.hadoop.hbase.client.ConnectionManager.HConnectionImplementation是如何加载我们需要的hbase数据表的信息的,我们看到hbase有个元数据表hbase:meta,这里hbase是namespace而meta是表名,我们自己创建的数据表的元数据信息都存储在这个元数据表hbase:meta中,第一次的时候会去元数据表hbase:meta中查找,找到后就加入缓存,第二次的时候直接从缓存获取我们的数据表的region信息
3.从分析源码中学到的对于hbase客户端的优化知识
- hbase客户端里传入hbase.client.write.buffer(默认2MB),加到客户端提交的缓存大小;
- hbase客户端提交采用批量提交,批量提交的List<Put>的size计算公式=hbase.client.write.buffer*2/Put大小,Put大小可通过put.heapSize()获取,以hbase.client.write.buffer=2097152,put.heapSize()=1320举例,最佳的批量提交记录大小=2*2097152/1320=3177;
- hbase客户端尽量采用多线程并发写
- hbase客户端所在机器性能要好,不然速度上不去
- 能接受关闭WAL的话尽量关闭,速度也会相应提升
4.hbase性能调研写入速度测试记录
相关推荐
赠送源代码:phoenix-core-4.7.0-HBase-1.1-sources.jar; 赠送Maven依赖信息文件:phoenix-core-4.7.0-HBase-1.1.pom; 包含翻译后的API文档:phoenix-core-4.7.0-HBase-1.1-javadoc-API文档-中文(简体)版.zip; ...
4. **配置HBase连接**:在Java项目中,你需要添加HBase的JAR包依赖,这通常包括HBase的客户端库,如`hbase-client.jar`。这些可以通过Maven或Gradle等构建工具管理。 5. **配置HBase配置文件**:创建一个名为`hbase...
"phoenix-5.0.0-HBase-2.0-client" 是一个针对Apache HBase数据库的Phoenix客户端库,主要用于通过SQL查询语句与HBase进行交互。这个版本的Phoenix客户端是为HBase 2.0版本设计和优化的,确保了与该版本HBase的兼容...
hbase-sdk是基于hbase-client和hbase-thrift的原生API封装的一款轻量级的HBase ORM框架。 针对HBase各版本API(1.x~2.x)间的差异,在其上剥离出了一层统一的抽象。并提供了以类SQL的方式来读写HBase表中的数据。对...
同时,启动器会根据这些配置动态地创建并管理HBase的客户端实例,使得开发者可以在代码中直接注入HBaseTemplate或者HBaseAdmin,进行数据的增删改查。 此外,`spring-boot-starter-hbase`还提供了对HBase的查询语句...
HBase 元数据修复工具包。 ①修改 jar 包中的application.properties,重点是 zookeeper.address、zookeeper.nodeParent、hdfs....③开始修复 `java -jar -Drepair.tableName=表名 hbase-meta-repair-hbase-2.0.2.jar`
Apache Phoenix是构建在HBase之上的关系型数据库层,作为内嵌的客户端JDBC驱动用以对HBase中的数据进行低延迟访问。Apache Phoenix会将用户编写的sql查询编译为一系列的scan操作,最终产生通用的JDBC结果集返回给...
phoenix-client-hbase-2.2-5.1.2.jar
hbase phoenix 客户端连接jdbc的jar包,SQuirreL SQL Client,DbVisualizer 等客户端连接hbase配置使用
在"apache-phoenix-4.14.0-HBase-1.2-src.tar.gz"这个压缩包中,我们主要会发现以下几个关键的知识点: 1. **Phoenix架构**:Phoenix采用了分层架构,包括客户端驱动、服务器端元数据服务、以及SQL编译器和执行器。...
1. **HBase Shell**:这是HBase自带的一个命令行接口,用户可以通过Java REPL(Read-Eval-Print Loop)与HBase交互。HBase Shell提供了创建表、删除表、插入数据、查询数据等基本操作,同时也支持复杂的条件查询和...
赠送源代码:phoenix-core-4.7.0-HBase-1.1-sources.jar; 赠送Maven依赖信息文件:phoenix-core-4.7.0-HBase-1.1.pom; 包含翻译后的API文档:phoenix-core-4.7.0-HBase-1.1-javadoc-API文档-中文(简体)-英语-对照...
这个“apache-kylin-3.0.2-bin-hbase1x.tar.gz”文件是Apache Kylin的3.0.2版本的二进制发行版,针对HBase 1.x版本进行了优化。下面我们将详细讨论Apache Kylin及其3.0.2版本的关键特性,以及与HBase的集成。 ...
在标题"apache-phoenix-4.8.1-HBase-1.2-bin.tar.gz"中,我们可以看到这是Apache Phoenix的4.8.1版本,它兼容HBase的1.2版本。这个压缩包是二进制发行版,通常包含了运行Phoenix所需的全部文件,包括JAR包、配置文件...
搭建pinpoint需要的hbase初始化脚本hbase-create.hbase
1. **自动配置**:Spring Boot的自动配置特性使得无需编写大量繁琐的配置代码,只需添加依赖,系统就能自动配置HBase的相关设置。 2. **客户端连接管理**:该组件提供了管理HBase客户端连接的工具,包括连接池的...
《Phoenix与HBase的深度解析:基于phoenix-hbase-2.4-5.1.2版本》 在大数据处理领域,Apache HBase和Phoenix是两个至关重要的组件。HBase作为一个分布式、列式存储的NoSQL数据库,为海量数据提供了高效、实时的访问...
这个"apache-phoenix-4.14.3-HBase-1.3-bin.tar.gz"文件是Phoenix的特定版本,针对HBase 1.3构建的二进制发行版。 1. **Apache Phoenix架构**:Phoenix主要由四部分组成:SQL解析器、元数据存储、优化器和执行器。...
phoenix-4.14.1-HBase-1.2-client.jar
毕业设计-基于java+HBase实现的手机数据备份系统(短信、联系人、重要文件).zip 基于HBase实现的手机数据备份系统,实现了手机关键信息的备份,如短信、联系人等。 包括服务器端(Server)和客户端(Client) Server...