1. Краткое введение
На данный момент (2019.04) HBase имеет две основные версии, 1.x и 2.x, API-интерфейсы Java двух версий различаются, некоторые методы в 1.x идентифицируются как в 2.x.@deprecatedустаревший. Поэтому для следующих примеров API я приведу версии 1.x и 2.x соответственно. Полный код можно найти в этом репозитории:
При этом версия используемого вами клиента должна совпадать с версией сервера.Если вы используете версию кода клиента 2.x для подключения к версии 1.x сервера, он выдаст ошибкуNoSuchColumnFamilyExceptionи другие исключения.
2. Базовое использование Java API 1.x
2.1 Создайте новый проект Maven и импортируйте зависимости проекта
Чтобы работать с HBase с помощью Java API, вам необходимо импортироватьhbase-client. выбрано здесьHBase Clientверсия1.2.0.
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>1.2.0</version>
</dependency>
2.2 Базовое использование API
public class HBaseUtils {
private static Connection connection;
static {
Configuration configuration = HBaseConfiguration.create();
configuration.set("hbase.zookeeper.property.clientPort", "2181");
// 如果是集群 则主机名用逗号分隔
configuration.set("hbase.zookeeper.quorum", "hadoop001");
try {
connection = ConnectionFactory.createConnection(configuration);
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 创建 HBase 表
*
* @param tableName 表名
* @param columnFamilies 列族的数组
*/
public static boolean createTable(String tableName, List<String> columnFamilies) {
try {
HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
if (admin.tableExists(tableName)) {
return false;
}
HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
columnFamilies.forEach(columnFamily -> {
HColumnDescriptor columnDescriptor = new HColumnDescriptor(columnFamily);
columnDescriptor.setMaxVersions(1);
tableDescriptor.addFamily(columnDescriptor);
});
admin.createTable(tableDescriptor);
} catch (IOException e) {
e.printStackTrace();
}
return true;
}
/**
* 删除 hBase 表
*
* @param tableName 表名
*/
public static boolean deleteTable(String tableName) {
try {
HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
// 删除表前需要先禁用表
admin.disableTable(tableName);
admin.deleteTable(tableName);
} catch (Exception e) {
e.printStackTrace();
}
return true;
}
/**
* 插入数据
*
* @param tableName 表名
* @param rowKey 唯一标识
* @param columnFamilyName 列族名
* @param qualifier 列标识
* @param value 数据
*/
public static boolean putRow(String tableName, String rowKey, String columnFamilyName, String qualifier,
String value) {
try {
Table table = connection.getTable(TableName.valueOf(tableName));
Put put = new Put(Bytes.toBytes(rowKey));
put.addColumn(Bytes.toBytes(columnFamilyName), Bytes.toBytes(qualifier), Bytes.toBytes(value));
table.put(put);
table.close();
} catch (IOException e) {
e.printStackTrace();
}
return true;
}
/**
* 插入数据
*
* @param tableName 表名
* @param rowKey 唯一标识
* @param columnFamilyName 列族名
* @param pairList 列标识和值的集合
*/
public static boolean putRow(String tableName, String rowKey, String columnFamilyName, List<Pair<String, String>> pairList) {
try {
Table table = connection.getTable(TableName.valueOf(tableName));
Put put = new Put(Bytes.toBytes(rowKey));
pairList.forEach(pair -> put.addColumn(Bytes.toBytes(columnFamilyName), Bytes.toBytes(pair.getKey()), Bytes.toBytes(pair.getValue())));
table.put(put);
table.close();
} catch (IOException e) {
e.printStackTrace();
}
return true;
}
/**
* 根据 rowKey 获取指定行的数据
*
* @param tableName 表名
* @param rowKey 唯一标识
*/
public static Result getRow(String tableName, String rowKey) {
try {
Table table = connection.getTable(TableName.valueOf(tableName));
Get get = new Get(Bytes.toBytes(rowKey));
return table.get(get);
} catch (IOException e) {
e.printStackTrace();
}
return null;
}
/**
* 获取指定行指定列 (cell) 的最新版本的数据
*
* @param tableName 表名
* @param rowKey 唯一标识
* @param columnFamily 列族
* @param qualifier 列标识
*/
public static String getCell(String tableName, String rowKey, String columnFamily, String qualifier) {
try {
Table table = connection.getTable(TableName.valueOf(tableName));
Get get = new Get(Bytes.toBytes(rowKey));
if (!get.isCheckExistenceOnly()) {
get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier));
Result result = table.get(get);
byte[] resultValue = result.getValue(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier));
return Bytes.toString(resultValue);
} else {
return null;
}
} catch (IOException e) {
e.printStackTrace();
}
return null;
}
/**
* 检索全表
*
* @param tableName 表名
*/
public static ResultScanner getScanner(String tableName) {
try {
Table table = connection.getTable(TableName.valueOf(tableName));
Scan scan = new Scan();
return table.getScanner(scan);
} catch (IOException e) {
e.printStackTrace();
}
return null;
}
/**
* 检索表中指定数据
*
* @param tableName 表名
* @param filterList 过滤器
*/
public static ResultScanner getScanner(String tableName, FilterList filterList) {
try {
Table table = connection.getTable(TableName.valueOf(tableName));
Scan scan = new Scan();
scan.setFilter(filterList);
return table.getScanner(scan);
} catch (IOException e) {
e.printStackTrace();
}
return null;
}
/**
* 检索表中指定数据
*
* @param tableName 表名
* @param startRowKey 起始 RowKey
* @param endRowKey 终止 RowKey
* @param filterList 过滤器
*/
public static ResultScanner getScanner(String tableName, String startRowKey, String endRowKey,
FilterList filterList) {
try {
Table table = connection.getTable(TableName.valueOf(tableName));
Scan scan = new Scan();
scan.setStartRow(Bytes.toBytes(startRowKey));
scan.setStopRow(Bytes.toBytes(endRowKey));
scan.setFilter(filterList);
return table.getScanner(scan);
} catch (IOException e) {
e.printStackTrace();
}
return null;
}
/**
* 删除指定行记录
*
* @param tableName 表名
* @param rowKey 唯一标识
*/
public static boolean deleteRow(String tableName, String rowKey) {
try {
Table table = connection.getTable(TableName.valueOf(tableName));
Delete delete = new Delete(Bytes.toBytes(rowKey));
table.delete(delete);
} catch (IOException e) {
e.printStackTrace();
}
return true;
}
/**
* 删除指定行的指定列
*
* @param tableName 表名
* @param rowKey 唯一标识
* @param familyName 列族
* @param qualifier 列标识
*/
public static boolean deleteColumn(String tableName, String rowKey, String familyName,
String qualifier) {
try {
Table table = connection.getTable(TableName.valueOf(tableName));
Delete delete = new Delete(Bytes.toBytes(rowKey));
delete.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(qualifier));
table.delete(delete);
table.close();
} catch (IOException e) {
e.printStackTrace();
}
return true;
}
}
2.3 Модульное тестирование
Протестируйте инкапсулированный выше API в режиме модульного тестирования.
public class HBaseUtilsTest {
private static final String TABLE_NAME = "class";
private static final String TEACHER = "teacher";
private static final String STUDENT = "student";
@Test
public void createTable() {
// 新建表
List<String> columnFamilies = Arrays.asList(TEACHER, STUDENT);
boolean table = HBaseUtils.createTable(TABLE_NAME, columnFamilies);
System.out.println("表创建结果:" + table);
}
@Test
public void insertData() {
List<Pair<String, String>> pairs1 = Arrays.asList(new Pair<>("name", "Tom"),
new Pair<>("age", "22"),
new Pair<>("gender", "1"));
HBaseUtils.putRow(TABLE_NAME, "rowKey1", STUDENT, pairs1);
List<Pair<String, String>> pairs2 = Arrays.asList(new Pair<>("name", "Jack"),
new Pair<>("age", "33"),
new Pair<>("gender", "2"));
HBaseUtils.putRow(TABLE_NAME, "rowKey2", STUDENT, pairs2);
List<Pair<String, String>> pairs3 = Arrays.asList(new Pair<>("name", "Mike"),
new Pair<>("age", "44"),
new Pair<>("gender", "1"));
HBaseUtils.putRow(TABLE_NAME, "rowKey3", STUDENT, pairs3);
}
@Test
public void getRow() {
Result result = HBaseUtils.getRow(TABLE_NAME, "rowKey1");
if (result != null) {
System.out.println(Bytes
.toString(result.getValue(Bytes.toBytes(STUDENT), Bytes.toBytes("name"))));
}
}
@Test
public void getCell() {
String cell = HBaseUtils.getCell(TABLE_NAME, "rowKey2", STUDENT, "age");
System.out.println("cell age :" + cell);
}
@Test
public void getScanner() {
ResultScanner scanner = HBaseUtils.getScanner(TABLE_NAME);
if (scanner != null) {
scanner.forEach(result -> System.out.println(Bytes.toString(result.getRow()) + "->" + Bytes
.toString(result.getValue(Bytes.toBytes(STUDENT), Bytes.toBytes("name")))));
scanner.close();
}
}
@Test
public void getScannerWithFilter() {
FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
SingleColumnValueFilter nameFilter = new SingleColumnValueFilter(Bytes.toBytes(STUDENT),
Bytes.toBytes("name"), CompareOperator.EQUAL, Bytes.toBytes("Jack"));
filterList.addFilter(nameFilter);
ResultScanner scanner = HBaseUtils.getScanner(TABLE_NAME, filterList);
if (scanner != null) {
scanner.forEach(result -> System.out.println(Bytes.toString(result.getRow()) + "->" + Bytes
.toString(result.getValue(Bytes.toBytes(STUDENT), Bytes.toBytes("name")))));
scanner.close();
}
}
@Test
public void deleteColumn() {
boolean b = HBaseUtils.deleteColumn(TABLE_NAME, "rowKey2", STUDENT, "age");
System.out.println("删除结果: " + b);
}
@Test
public void deleteRow() {
boolean b = HBaseUtils.deleteRow(TABLE_NAME, "rowKey2");
System.out.println("删除结果: " + b);
}
@Test
public void deleteTable() {
boolean b = HBaseUtils.deleteTable(TABLE_NAME);
System.out.println("删除结果: " + b);
}
}
3. Базовое использование Java API 2.x
3.1 Создайте новый проект Maven и импортируйте зависимости проекта
выбрано здесьHBase Clientверсия самая последняя2.1.4.
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>2.1.4</version>
</dependency>
3.2 Базовое использование API
По сравнению с 1.x, в версии 2.x некоторые методы устарели, а новые альтернативные методы будут указаны в исходном коде для устаревших методов, например, при создании таблицы в 2.x:HTableDescriptorиHColumnDescriptorи другие классы помечаются как устаревшие и заменяются использованиемTableDescriptorBuilderиColumnFamilyDescriptorBuilderдля определения таблиц и семейств столбцов.
Ниже приведен пример использования версии HBase 2.x API Java:
public class HBaseUtils {
private static Connection connection;
static {
Configuration configuration = HBaseConfiguration.create();
configuration.set("hbase.zookeeper.property.clientPort", "2181");
// 如果是集群 则主机名用逗号分隔
configuration.set("hbase.zookeeper.quorum", "hadoop001");
try {
connection = ConnectionFactory.createConnection(configuration);
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 创建 HBase 表
*
* @param tableName 表名
* @param columnFamilies 列族的数组
*/
public static boolean createTable(String tableName, List<String> columnFamilies) {
try {
HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
if (admin.tableExists(TableName.valueOf(tableName))) {
return false;
}
TableDescriptorBuilder tableDescriptor = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName));
columnFamilies.forEach(columnFamily -> {
ColumnFamilyDescriptorBuilder cfDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(columnFamily));
cfDescriptorBuilder.setMaxVersions(1);
ColumnFamilyDescriptor familyDescriptor = cfDescriptorBuilder.build();
tableDescriptor.setColumnFamily(familyDescriptor);
});
admin.createTable(tableDescriptor.build());
} catch (IOException e) {
e.printStackTrace();
}
return true;
}
/**
* 删除 hBase 表
*
* @param tableName 表名
*/
public static boolean deleteTable(String tableName) {
try {
HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
// 删除表前需要先禁用表
admin.disableTable(TableName.valueOf(tableName));
admin.deleteTable(TableName.valueOf(tableName));
} catch (Exception e) {
e.printStackTrace();
}
return true;
}
/**
* 插入数据
*
* @param tableName 表名
* @param rowKey 唯一标识
* @param columnFamilyName 列族名
* @param qualifier 列标识
* @param value 数据
*/
public static boolean putRow(String tableName, String rowKey, String columnFamilyName, String qualifier,
String value) {
try {
Table table = connection.getTable(TableName.valueOf(tableName));
Put put = new Put(Bytes.toBytes(rowKey));
put.addColumn(Bytes.toBytes(columnFamilyName), Bytes.toBytes(qualifier), Bytes.toBytes(value));
table.put(put);
table.close();
} catch (IOException e) {
e.printStackTrace();
}
return true;
}
/**
* 插入数据
*
* @param tableName 表名
* @param rowKey 唯一标识
* @param columnFamilyName 列族名
* @param pairList 列标识和值的集合
*/
public static boolean putRow(String tableName, String rowKey, String columnFamilyName, List<Pair<String, String>> pairList) {
try {
Table table = connection.getTable(TableName.valueOf(tableName));
Put put = new Put(Bytes.toBytes(rowKey));
pairList.forEach(pair -> put.addColumn(Bytes.toBytes(columnFamilyName), Bytes.toBytes(pair.getKey()), Bytes.toBytes(pair.getValue())));
table.put(put);
table.close();
} catch (IOException e) {
e.printStackTrace();
}
return true;
}
/**
* 根据 rowKey 获取指定行的数据
*
* @param tableName 表名
* @param rowKey 唯一标识
*/
public static Result getRow(String tableName, String rowKey) {
try {
Table table = connection.getTable(TableName.valueOf(tableName));
Get get = new Get(Bytes.toBytes(rowKey));
return table.get(get);
} catch (IOException e) {
e.printStackTrace();
}
return null;
}
/**
* 获取指定行指定列 (cell) 的最新版本的数据
*
* @param tableName 表名
* @param rowKey 唯一标识
* @param columnFamily 列族
* @param qualifier 列标识
*/
public static String getCell(String tableName, String rowKey, String columnFamily, String qualifier) {
try {
Table table = connection.getTable(TableName.valueOf(tableName));
Get get = new Get(Bytes.toBytes(rowKey));
if (!get.isCheckExistenceOnly()) {
get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier));
Result result = table.get(get);
byte[] resultValue = result.getValue(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier));
return Bytes.toString(resultValue);
} else {
return null;
}
} catch (IOException e) {
e.printStackTrace();
}
return null;
}
/**
* 检索全表
*
* @param tableName 表名
*/
public static ResultScanner getScanner(String tableName) {
try {
Table table = connection.getTable(TableName.valueOf(tableName));
Scan scan = new Scan();
return table.getScanner(scan);
} catch (IOException e) {
e.printStackTrace();
}
return null;
}
/**
* 检索表中指定数据
*
* @param tableName 表名
* @param filterList 过滤器
*/
public static ResultScanner getScanner(String tableName, FilterList filterList) {
try {
Table table = connection.getTable(TableName.valueOf(tableName));
Scan scan = new Scan();
scan.setFilter(filterList);
return table.getScanner(scan);
} catch (IOException e) {
e.printStackTrace();
}
return null;
}
/**
* 检索表中指定数据
*
* @param tableName 表名
* @param startRowKey 起始 RowKey
* @param endRowKey 终止 RowKey
* @param filterList 过滤器
*/
public static ResultScanner getScanner(String tableName, String startRowKey, String endRowKey,
FilterList filterList) {
try {
Table table = connection.getTable(TableName.valueOf(tableName));
Scan scan = new Scan();
scan.withStartRow(Bytes.toBytes(startRowKey));
scan.withStopRow(Bytes.toBytes(endRowKey));
scan.setFilter(filterList);
return table.getScanner(scan);
} catch (IOException e) {
e.printStackTrace();
}
return null;
}
/**
* 删除指定行记录
*
* @param tableName 表名
* @param rowKey 唯一标识
*/
public static boolean deleteRow(String tableName, String rowKey) {
try {
Table table = connection.getTable(TableName.valueOf(tableName));
Delete delete = new Delete(Bytes.toBytes(rowKey));
table.delete(delete);
} catch (IOException e) {
e.printStackTrace();
}
return true;
}
/**
* 删除指定行指定列
*
* @param tableName 表名
* @param rowKey 唯一标识
* @param familyName 列族
* @param qualifier 列标识
*/
public static boolean deleteColumn(String tableName, String rowKey, String familyName,
String qualifier) {
try {
Table table = connection.getTable(TableName.valueOf(tableName));
Delete delete = new Delete(Bytes.toBytes(rowKey));
delete.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(qualifier));
table.delete(delete);
table.close();
} catch (IOException e) {
e.printStackTrace();
}
return true;
}
}
4. Правильно подключите Hbase
В приведенном выше коде соединение Connection инициализируется при загрузке класса, а последующие методы предназначены для повторного использования соединения. В настоящее время мы можем подумать, можем ли мы использовать настраиваемый пул соединений для повышения производительности? На самом деле в этом нет необходимости.
Во-первых, официальныйConnectionИнструкции по применению следующие:
Connection Pooling For applications which require high-end multithreaded
access (e.g., web-servers or application servers that may serve many
application threads in a single JVM), you can pre-create a Connection,
as shown in the following example:
对于高并发多线程访问的应用程序（例如，在单个 JVM 中存在的为多个线程服务的 Web 服务器或应用程序服务器），
您只需要预先创建一个 Connection。例子如下：
// Create a connection to the cluster.
Configuration conf = HBaseConfiguration.create();
try (Connection connection = ConnectionFactory.createConnection(conf);
Table table = connection.getTable(TableName.valueOf(tablename))) {
// use table as needed, the table returned is lightweight
}
Это можно использовать, потому что Connection не является простым соединением через сокет,документация по интерфейсуВыражение для Connection in:
A cluster connection encapsulating lower level individual connections to actual servers and a
connection to zookeeper. Connections are instantiated through the ConnectionFactory class.
The lifecycle of the connection is managed by the caller, who has to close() the connection
to release the resources.
Connection 是一个集群连接，封装了与多台服务器（Matser/Region Server）的底层连接以及与 zookeeper 的连接。
连接通过 ConnectionFactory 类实例化。连接的生命周期由调用者管理，调用者必须使用 close() 关闭连接以释放资源。
Эти соединения инкапсулированы, поскольку клиентам HBase необходимо подключаться к трем разным ролям службы:
-
Zookeeper: в основном используется для получения
metaИнформация о местонахождении стола, информация Мастера; - HBase Master: В основном используется для выполнения некоторых операций интерфейса HBaseAdmin, таких как создание таблиц и т. д.;
- HBase RegionServer: Используется для чтения и записи данных.
Соответствие между объектом Connection и фактическим соединением Socket выглядит следующим образом:
Две приведенные выше фотографии взяты из блога:Правильная поза для подключения к HBase
В клиентском коде HBase соединение Socket действительно соответствуетRpcConnectionобъект. Использование HBasePoolMapЭта структура данных для хранения соединения между клиентом и сервером HBase.PoolMapвнутри естьConcurrentHashMapэкземпляр, ключ которогоConnectionId(инкапсулирует адрес сервера и билет пользователя), значение представляет собойRpcConnectionПул ресурсов объекта. Когда HBase необходимо подключиться к серверу, он сначалаConnectionIdНайдите соответствующий пул соединений, а затем возьмите объект соединения из пула соединений.
@InterfaceAudience.Private
public class PoolMap<K, V> implements Map<K, V> {
private PoolType poolType;
private int poolMaxSize;
private Map<K, Pool<V>> pools = new ConcurrentHashMap<>();
public PoolMap(PoolType poolType) {
this.poolType = poolType;
}
.....
В HBase есть три реализации пулов ресурсов, а именноReusable,RoundRobinиThreadLocal. Конкретная реализация может бытьhbase.client.ipc.pool.typeУказанный элемент конфигурации, по умолчаниюReusable. Размер пула соединений также можно установить с помощьюhbase.client.ipc.pool.sizeВ элементе конфигурации указано, по умолчанию 1, то есть 1 подключение на сервер. Это также может быть достигнуто путем изменения конфигурации:
config.set("hbase.client.ipc.pool.type",...);
config.set("hbase.client.ipc.pool.size",...);
connection = ConnectionFactory.createConnection(config);
Из этого видно, что класс Connection в HBase уже реализовал функцию управления соединением, поэтому нам не нужно делать дополнительное управление на Connection.
Кроме того, Connection является потокобезопасным, но Table и Admin не являются потокобезопасными, поэтому правильным подходом является совместное использование одного объекта Connection для одного процесса и использование отдельных объектов Table и Admin в разных потоках. Получить работу таблицы и администратораgetTable()иgetAdmin()Оба легкие, поэтому вам не нужно беспокоиться о потреблении производительности, и рекомендуется отображать вызов после завершения использования.close()метод их закрытия.
использованная литература
Другие статьи серии больших данных можно найти в проекте с открытым исходным кодом GitHub.:Руководство для начинающих по большим данным