第三章：HBase 客户端 API 与编程

文档摘要

第三章：HBase 客户端 API 与编程第三章：HBase 客户端 API 与编程 3.1 HBase 客户端 API 概览 HBase 客户端 API 提供了一组 Java 类和接口，用于与 HBase 集群进行交互。主要的类包括： : 用于配置 HBase 客户端，指定 HBase 集群的地址、端口和其他相关参数。 : 代表与 HBase 集群的连接。它是线程安全的，应该被尽可能重用。 : 提供管理 HBase 集群的功能，例如创建、删除和修改表。 : 代表 HBase 中的一个表，用于执行数据操作，例如插入、检索、更新和删除。 : 用于从表中检索单行数据。 : 用于向表中插入或更新数据。 : 用于从表中删除数据。 : 用于扫描表中的多行数据。

第三章：HBase 客户端 API 与编程

3.1 HBase 客户端 API 概览

HBase 客户端 API 提供了一组 Java 类和接口，用于与 HBase 集群进行交互。主要的类包括：

Configuration: 用于配置 HBase 客户端，指定 HBase 集群的地址、端口和其他相关参数。
Connection: 代表与 HBase 集群的连接。它是线程安全的，应该被尽可能重用。
Admin: 提供管理 HBase 集群的功能，例如创建、删除和修改表。
Table: 代表 HBase 中的一个表，用于执行数据操作，例如插入、检索、更新和删除。
Get: 用于从表中检索单行数据。
Put: 用于向表中插入或更新数据。
Delete: 用于从表中删除数据。
Scan: 用于扫描表中的多行数据。
Result: 代表从 HBase 中检索到的数据。
ResultScanner: 用于迭代 Scan 操作返回的结果。

3.2 连接到 HBase 集群

首先，我们需要建立与 HBase 集群的连接。以下代码演示了如何使用 Configuration 和 ConnectionFactory 来创建连接：


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import java.io.IOException;
public class HBaseConnection {
    public static void main(String[] args) throws IOException {
        // 1. 创建 HBase 配置对象
        Configuration conf = HBaseConfiguration.create();
        //  配置 HBase 集群地址 (根据实际情况修改)
        conf.set("hbase.zookeeper.quorum", "localhost");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        // 2. 建立与 HBase 的连接
        try (Connection connection = ConnectionFactory.createConnection(conf)) {
            System.out.println("Successfully connected to HBase!");
        } catch (IOException e) {
            System.err.println("Failed to connect to HBase: " + e.getMessage());
        }
    }
}

代码解释：

创建配置对象： 使用 HBaseConfiguration.create() 创建一个默认的 HBase 配置对象。
配置连接参数： 设置 hbase.zookeeper.quorum 和 hbase.zookeeper.property.clientPort 属性，指定 ZooKeeper 集群的地址和端口。ZooKeeper 用于管理 HBase 集群的元数据。请根据你的 HBase 集群配置修改这些值。
建立连接： 使用 ConnectionFactory.createConnection(conf) 创建一个 Connection 对象。 try-with-resources 语句确保连接在使用后会被自动关闭。
异常处理： 使用 try-catch 块捕获可能发生的 IOException，并在控制台打印错误信息。

3.3 表管理：创建、删除和修改

Admin 接口用于管理 HBase 表。以下代码演示了如何创建、删除表。


import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import java.io.IOException;
public class HBaseTableManagement {
    public static void main(String[] args) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "localhost");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        try (Connection connection = ConnectionFactory.createConnection(conf);
             Admin admin = connection.getAdmin()) {
            // 1. 定义表名
            TableName tableName = TableName.valueOf("mytable");
            // 2. 创建表
            if (!admin.tableExists(tableName)) {
                HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);
                tableDescriptor.addFamily(new HColumnDescriptor("cf1")); // 添加列族
                tableDescriptor.addFamily(new HColumnDescriptor("cf2"));
                admin.createTable(tableDescriptor);
                System.out.println("Table 'mytable' created successfully.");
            } else {
                System.out.println("Table 'mytable' already exists.");
            }
            // 3. 删除表
            // 禁用表
            if (admin.tableExists(tableName)) {
                if(admin.isTableEnabled(tableName)){
                    admin.disableTable(tableName);
                }
                // 删除表
                admin.deleteTable(tableName);
                System.out.println("Table 'mytable' deleted successfully.");
            } else {
                System.out.println("Table 'mytable' does not exist.");
            }
        } catch (IOException e) {
            System.err.println("Error during table management: " + e.getMessage());
        }
    }
}

代码解释：

获取 Admin 实例： 通过 connection.getAdmin() 获取 Admin 实例。
定义表名： 使用 TableName.valueOf() 创建一个 TableName 对象，表示要操作的表名。
创建表：
- 首先，使用 admin.tableExists() 检查表是否已存在。
- 如果表不存在，创建一个 HTableDescriptor 对象，用于描述表的结构。
- 使用 tableDescriptor.addFamily() 添加一个或多个列族。每个表必须至少有一个列族。
- 使用 admin.createTable() 创建表。
删除表：
- 首先，使用 admin.tableExists() 检查表是否存在。
- 如果表存在，首先使用 admin.disableTable() 禁用该表。在删除表之前必须禁用它。
- 然后，使用 admin.deleteTable() 删除该表。

3.4 数据操作：插入、检索、更新和删除

Table 接口用于执行数据操作。以下代码演示了如何插入、检索、更新和删除数据。


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import java.io.IOException;
public class HBaseDataOperations {
    public static void main(String[] args) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "localhost");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        TableName tableName = TableName.valueOf("mytable");
        try (Connection connection = ConnectionFactory.createConnection(conf);
             Table table = connection.getTable(tableName)) {
            // 1. 插入数据 (Put)
            Put put = new Put(Bytes.toBytes("row1")); // Row key
            put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("name"), Bytes.toBytes("John"));
            put.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("age"), Bytes.toBytes("30"));
            table.put(put);
            System.out.println("Data inserted for row 'row1'.");
            // 2. 检索数据 (Get)
            Get get = new Get(Bytes.toBytes("row1"));
            Result result = table.get(get);
            byte[] nameBytes = result.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("name"));
            byte[] ageBytes = result.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("age"));
            String name = Bytes.toString(nameBytes);
            String age = Bytes.toString(ageBytes);
            System.out.println("Name: " + name + ", Age: " + age);
            // 3. 更新数据 (Put - overwrite)
            Put update = new Put(Bytes.toBytes("row1"));
            update.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("age"), Bytes.toBytes("35")); // 更新年龄
            table.put(update);
            System.out.println("Data updated for row 'row1'.");
            // 4. 再次检索，验证更新
            Result updatedResult = table.get(get);
            byte[] updatedAgeBytes = updatedResult.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("age"));
            String updatedAge = Bytes.toString(updatedAgeBytes);
            System.out.println("Updated Age: " + updatedAge);
            // 5. 删除数据 (Delete)
            Delete delete = new Delete(Bytes.toBytes("row1"));
            delete.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("age")); // 删除 'age' 列
            table.delete(delete);
            System.out.println("Column 'age' deleted from row 'row1'.");
            // 6. 再次检索，验证删除
            Result deletedResult = table.get(get);
            byte[] deletedAgeBytes = deletedResult.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("age"));
            if (deletedAgeBytes == null) {
                System.out.println("Column 'age' is indeed deleted.");
            } else {
                System.out.println("Column 'age' was not deleted.");
            }
        } catch (IOException e) {
            System.err.println("Error during data operations: " + e.getMessage());
        }
    }
}

代码解释：

获取 Table 实例： 通过 connection.getTable(tableName) 获取 Table 实例。
插入数据 (Put)：
- 创建一个 Put 对象，指定要插入的行键。
- 使用 put.addColumn() 方法添加要插入的列。需要指定列族、列限定符和值。
- 使用 table.put() 方法将数据插入到表中。
检索数据 (Get)：
- 创建一个 Get 对象，指定要检索的行键。
- 使用 table.get() 方法检索数据。
- 使用 result.getValue() 方法获取指定列族和列限定符的值。
- 使用 Bytes.toString() 将字节数组转换为字符串。
更新数据 (Put - overwrite)：
- 更新数据与插入数据使用相同的方法，即使用 Put 对象。
- 如果指定的行键和列已存在，则 Put 操作将覆盖现有值。
删除数据 (Delete)：
- 创建一个 Delete 对象，指定要删除的行键。
- 可以使用 delete.addColumn() 删除特定的列，也可以使用 delete.addFamily() 删除整个列族。
- 使用 table.delete() 方法删除数据。

3.5 扫描数据：Scan 和 ResultScanner

Scan 用于扫描表中的多行数据。ResultScanner 用于迭代 Scan 操作返回的结果。


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import java.io.IOException;
public class HBaseScan {
    public static void main(String[] args) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "localhost");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        TableName tableName = TableName.valueOf("mytable");
        try (Connection connection = ConnectionFactory.createConnection(conf);
             Table table = connection.getTable(tableName)) {
            // 1. 创建 Scan 对象
            Scan scan = new Scan();
            // 2. 设置 Scan 参数 (可选)
            //scan.setStartRow(Bytes.toBytes("row1")); // 设置起始行键
            //scan.setStopRow(Bytes.toBytes("row5"));  // 设置结束行键
            //scan.addFamily(Bytes.toBytes("cf1"));   // 设置要扫描的列族
            // 3. 执行 Scan 操作
            try (ResultScanner scanner = table.getScanner(scan)) {
                // 4. 迭代结果
                for (Result result : scanner) {
                    byte[] rowKey = result.getRow();
                    byte[] nameBytes = result.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("name"));
                    byte[] ageBytes = result.getValue(Bytes.toBytes("cf1"), Bytes.toBytes("age"));
                    String name = (nameBytes != null) ? Bytes.toString(nameBytes) : "N/A";
                    String age = (ageBytes != null) ? Bytes.toString(ageBytes) : "N/A";
                    System.out.println("Row Key: " + Bytes.toString(rowKey) + ", Name: " + name + ", Age: " + age);
                }
            }
        } catch (IOException e) {
            System.err.println("Error during scan operation: " + e.getMessage());
        }
    }
}

代码解释：

创建 Scan 对象： 创建一个 Scan 对象，用于定义扫描操作。
设置 Scan 参数 (可选)：
- scan.setStartRow(): 设置扫描的起始行键。
- scan.setStopRow(): 设置扫描的结束行键。扫描结果将不包括该行。
- scan.addFamily(): 设置要扫描的列族。如果不指定，将扫描所有列族。
执行 Scan 操作： 使用 table.getScanner(scan) 执行扫描操作，并返回一个 ResultScanner 对象。
迭代结果： 使用 for 循环迭代 ResultScanner 对象，获取每个 Result 对象。
- 使用 result.getRow() 获取行键。
- 使用 result.getValue() 获取指定列族和列限定符的值。

3.6 过滤器：控制扫描结果

HBase 过滤器允许你根据特定的条件过滤扫描结果。 HBase 提供了多种内置过滤器，例如：

RowFilter: 基于行键进行过滤。
FamilyFilter: 基于列族进行过滤。
QualifierFilter: 基于列限定符进行过滤。
ValueFilter: 基于单元格值进行过滤。
PrefixFilter: 匹配具有特定前缀的行键。

以下代码演示了如何使用 RowFilter 过滤扫描结果：


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.BinaryPrefixComparator;
import org.apache.hadoop.hbase.util.Bytes;
import java.io.IOException;
public class HBaseFilter {
    public static void main(String[] args) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "localhost");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        TableName tableName = TableName.valueOf("mytable");
        try (Connection connection = ConnectionFactory.createConnection(conf);
             Table table = connection.getTable(tableName)) {
            // 1. 创建 Scan 对象
            Scan scan = new Scan();
            // 2. 创建 RowFilter，过滤以 "row" 开头的行键
            RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, new BinaryPrefixComparator(Bytes.toBytes("row")));
            // 3. 将过滤器添加到 Scan 对象
            scan.setFilter(rowFilter);
            // 4. 执行 Scan 操作
            try (ResultScanner scanner = table.getScanner(scan)) {
                // 5. 迭代结果
                for (Result result : scanner) {
                    byte[] rowKey = result.getRow();
                    System.out.println("Row Key: " + Bytes.toString(rowKey));
                }
            }
        } catch (IOException e) {
            System.err.println("Error during scan operation: " + e.getMessage());
        }
    }
}

代码解释：

创建 RowFilter： 创建一个 RowFilter 对象，使用 CompareOp.EQUAL 和 BinaryPrefixComparator 来匹配以 "row" 开头的行键。
将过滤器添加到 Scan 对象： 使用 scan.setFilter() 方法将过滤器添加到 Scan 对象。

3.7 代码实践总结

本章介绍了 HBase 客户端 API 的核心概念和用法。通过学习如何连接到 HBase 集群、管理表、执行数据操作以及使用扫描和过滤器，你已经掌握了 HBase 客户端编程的基础。

Graph TD 图示：

图示解释：

Configuration 用于配置 HBase 连接，通过 ConnectionFactory 创建 Connection 对象。
Connection 对象可以获取 Admin 和 Table 对象。
Admin 对象用于管理表，例如创建和删除表。
Table 对象用于执行数据操作，例如插入、检索、更新和删除数据，以及扫描数据。
Scan 操作返回 ResultScanner 对象，用于迭代扫描结果。
ResultScanner 迭代返回 Result 对象，代表一行数据。
Result 对象可以获取指定列族和列限定符的值。

通过熟练掌握这些 API，你将能够构建强大的应用程序，充分利用 HBase 的强大功能。在实际应用中，请务必根据你的具体需求选择合适的 API 和参数，并进行充分的测试。

作者与出处

原作者：灏天文库

来源：灏天文库

原文： https://aiknowledge.cn/collection/151-hbase%E5%88%86%E5%B8%83%E5%BC%8F%E6%95%B0%E6%8D%AE%E5%BA%93%E5%85%A5%E9%97%A8%E4%B8%8E%E5%AE%9E%E6%88%98

整理：灏天文库整理

由灏天文库平台收录，内容或由平台用户上传，仅供学习交流