HBase客户端API

HBase Java API 实战：连接、增删改查、过滤器、计数器，一套代码全搞定

HBase 的 Java API 不复杂，核心就几个类：

类	管什么
Connection	连集群（线程安全，全局一个）
Admin	建表、删表、改表结构
Table	增删改查数据
Put / Get / Delete / Scan	各操作的参数封装

环境准备与连接管理

在使用 HBase API 前，需确保项目引入 HBase 依赖（以 Maven 为例），并正确配置 HBase 连接信息（hbase-site.xml 需放在项目 classpath 下）。

依赖引入

<dependency>  
    <groupId>org.apache.hbase</groupId>  
    <artifactId>hbase-client</artifactId>  
    <version>2.2.7</version>  <!-- 与 HBase 集群版本一致 -->  
</dependency>

连接初始化：全局一个 Connection

Connection 是线程安全的，整个应用生命周期里用一个就行，不要每次操作都 new。

import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.TableName;

public class HBaseClient {
    // 全局连接（线程安全，单例）
    private static Connection connection;
    private static Table table;
    
    static {
        try {
            // 自动加载 classpath 下的 hbase-site.xml
            connection = ConnectionFactory.createConnection();
            table = connection.getTable(TableName.valueOf("test"));
        } catch (IOException e) {
            throw new RuntimeException("HBase 连接初始化失败", e);
        }
    }
    
    // 如果操作多张表，用这个方法获取 Table 实例
    public static Table getTable(String tableName) throws IOException {
        return connection.getTable(TableName.valueOf(tableName));
    }
}

关键点： ConnectionFactory.createConnection() 会自动读取 hbase-site.xml，把它放在 src/main/resources 下就行。

插入数据（Put）

public static void putData(String rowKey, String family, String qualifier, String value) 
        throws IOException {
    Put put = new Put(Bytes.toBytes(rowKey));
    put.addColumn(
        Bytes.toBytes(family),
        Bytes.toBytes(qualifier),
        Bytes.toBytes(value)
    );
    table.put(put);
}

调用：

putData("user_001", "info", "name", "张三");
putData("user_001", "info", "age", "25");
putData("user_001", "log", "login_time", "2024-01-15 08:00:00");

批量插入（性能更好）：

public static void batchPut(List<Put> puts) throws IOException {
    table.put(puts);
}

查询单行（Get）

public static String getData(String rowKey, String family, String qualifier) 
        throws IOException {
    Get get = new Get(Bytes.toBytes(rowKey));
    get.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
    
    Result result = table.get(get);
    byte[] value = result.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier));
    return value == null ? null : Bytes.toString(value);
}

调用：

String name = getData("user_001", "info", "name");
System.out.println(name);  // 张三

查整行（不加列过滤）：

Get get = new Get(Bytes.toBytes(rowKey));
Result result = table.get(get);
// 遍历所有列
for (Cell cell : result.listCells()) {
    String family = Bytes.toString(CellUtil.cloneFamily(cell));
    String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
    String value = Bytes.toString(CellUtil.cloneValue(cell));
    System.out.println(family + ":" + qualifier + " = " + value);
}

删除数据（Delete）

// 删除单列
public static void deleteColumn(String rowKey, String family, String qualifier) 
        throws IOException {
    Delete delete = new Delete(Bytes.toBytes(rowKey));
    delete.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
    table.delete(delete);
}

// 删除整行
public static void deleteRow(String rowKey) throws IOException {
    Delete delete = new Delete(Bytes.toBytes(rowKey));
    table.delete(delete);
}

批量扫描（Scan）

public static void scanData(String family, String qualifier, int limit) 
        throws IOException {
    Scan scan = new Scan();
    scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
    
    // 分页：只返回前 limit 行
    if (limit > 0) {
        scan.setFilter(new PageFilter(limit));
    }
    
    try (ResultScanner scanner = table.getScanner(scan)) {
        for (Result result : scanner) {
            String rowKey = Bytes.toString(result.getRow());
            String value = Bytes.toString(
                result.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier))
            );
            System.out.println("RowKey: " + rowKey + ", Value: " + value);
        }
    }
}

范围扫描：

Scan scan = new Scan();
scan.withStartRow(Bytes.toBytes("user_001"));
scan.withStopRow(Bytes.toBytes("user_100"));

计数器（原子累加）

public static long increment(String rowKey, String family, String qualifier, long delta) 
        throws IOException {
    return table.incrementColumnValue(
        Bytes.toBytes(rowKey),
        Bytes.toBytes(family),
        Bytes.toBytes(qualifier),
        delta // 步长，可正可负
    );
}

// 查看当前值（delta=0 不改变值）
public static long getCounter(String rowKey, String family, String qualifier) 
        throws IOException {
    return table.incrementColumnValue(
        Bytes.toBytes(rowKey),
        Bytes.toBytes(family),
        Bytes.toBytes(qualifier),
        0
    );
}

调用：

// 累加 +1
increment("ad_001", "stats", "impression", 1);
// 累加 +5
increment("ad_001", "stats", "impression", 5);
// 查当前值
long count = getCounter("ad_001", "stats", "impression");
System.out.println("曝光量: " + count);

计数器是原子操作，多个客户端同时累加不会乱。

资源管理：用 try-with-resources

Table 和 ResultScanner 需要关闭，用 try-with-resources 最省事：

//  推荐
try (Table table = connection.getTable(TableName.valueOf("test"))) {
    // 操作 table
} catch (IOException e) {
    // 处理异常
}

//  不推荐：手动 close，容易漏
Table table = connection.getTable(...);
// 操作...
table.close();  // 万一中间抛异常，close 不执行

全局 Connection 在应用关闭时释放：

// 应用关闭时调用
public static void close() {
    try {
        if (table != null) table.close();
        if (connection != null) connection.close();
    } catch (IOException e) {
        e.printStackTrace();
    }
}

完整代码结构

import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.filter.PageFilter;
import java.io.IOException;

public class HBaseClient {
    private static Connection connection;
    
    static {
        try {
            connection = ConnectionFactory.createConnection();
        } catch (IOException e) {
            throw new RuntimeException("HBase 连接失败", e);
        }
    }
    
    // 获取 Table 实例（每次用 try-with-resources）
    public static Table getTable(String tableName) throws IOException {
        return connection.getTable(TableName.valueOf(tableName));
    }
    
    // ---- 增删改查 ----
    public static void putData(String tableName, String rowKey, String family, 
                               String qualifier, String value) throws IOException {
        try (Table table = getTable(tableName)) {
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier), Bytes.toBytes(value));
            table.put(put);
        }
    }
    
    public static String getData(String tableName, String rowKey, String family, 
                                 String qualifier) throws IOException {
        try (Table table = getTable(tableName)) {
            Get get = new Get(Bytes.toBytes(rowKey));
            get.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
            Result result = table.get(get);
            byte[] value = result.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier));
            return value == null ? null : Bytes.toString(value);
        }
    }
    
    public static void deleteData(String tableName, String rowKey, String family, 
                                  String qualifier) throws IOException {
        try (Table table = getTable(tableName)) {
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            delete.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
            table.delete(delete);
        }
    }
    
    public static long increment(String tableName, String rowKey, String family, 
                                 String qualifier, long delta) throws IOException {
        try (Table table = getTable(tableName)) {
            return table.incrementColumnValue(
                Bytes.toBytes(rowKey),
                Bytes.toBytes(family),
                Bytes.toBytes(qualifier),
                delta
            );
        }
    }
    
    public static void close() {
        try {
            if (connection != null) connection.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    
    public static void main(String[] args) throws IOException {
        String tableName = "test";
        
        // 插入
        putData(tableName, "user_001", "info", "name", "张三");
        
        // 查询
        String name = getData(tableName, "user_001", "info", "name");
        System.out.println("姓名: " + name);
        
        // 计数器
        long count = increment(tableName, "ad_001", "stats", "impression", 1);
        System.out.println("计数: " + count);
        
        // 清理
        close();
    }
}