Apache HBase 技术文档

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;

public class HBaseConnection {
    private static Connection connection;

    public static Connection getConnection() throws IOException {
        if (connection == null) {
            Configuration config = HBaseConfiguration.create();
            config.set("hbase.zookeeper.quorum", "localhost");
            config.set("hbase.zookeeper.property.clientPort", "2181");
            connection = ConnectionFactory.createConnection(config);
        }
        return connection;
    }
}

import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;

public class TableManager {

    // 创建表
    public void createTable(String tableName, String... columnFamilies)
            throws IOException {
        try (Connection conn = HBaseConnection.getConnection();
             Admin admin = conn.getAdmin()) {

            TableDescriptorBuilder builder = TableDescriptorBuilder
                .newBuilder(TableName.valueOf(tableName));

            for (String cf : columnFamilies) {
                ColumnFamilyDescriptorBuilder cfBuilder =
                    ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cf));
                builder.setColumnFamily(cfBuilder.build());
            }

            admin.createTable(builder.build());
            System.out.println("表 " + tableName + " 创建成功");
        }
    }

    // 删除表
    public void deleteTable(String tableName) throws IOException {
        try (Connection conn = HBaseConnection.getConnection();
             Admin admin = conn.getAdmin()) {

            TableName table = TableName.valueOf(tableName);
            admin.disableTable(table);
            admin.deleteTable(table);
            System.out.println("表 " + tableName + " 删除成功");
        }
    }
}

import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;

public class DataOperations {

    // 插入单条数据
    public void putData(String tableName, String rowKey,
                       String columnFamily, String column, String value)
            throws IOException {
        try (Connection conn = HBaseConnection.getConnection();
             Table table = conn.getTable(TableName.valueOf(tableName))) {

            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(columnFamily),
                         Bytes.toBytes(column),
                         Bytes.toBytes(value));

            table.put(put);
            System.out.println("数据插入成功");
        }
    }

    // 批量插入数据
    public void batchPut(String tableName, List<Put> puts)
            throws IOException {
        try (Connection conn = HBaseConnection.getConnection();
             Table table = conn.getTable(TableName.valueOf(tableName))) {

            table.put(puts);
            System.out.println("批量插入 " + puts.size() + " 条数据");
        }
    }
}

import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;

public class DataQuery {

    // 根据行键查询
    public void getByRowKey(String tableName, String rowKey)
            throws IOException {
        try (Connection conn = HBaseConnection.getConnection();
             Table table = conn.getTable(TableName.valueOf(tableName))) {

            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = table.get(get);

            if (!result.isEmpty()) {
                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));
                    long timestamp = cell.getTimestamp();

                    System.out.println("列族: " + family +
                                     ", 列: " + qualifier +
                                     ", 值: " + value +
                                     ", 时间戳: " + timestamp);
                }
            }
        }
    }

    // 扫描表数据
    public void scanTable(String tableName) throws IOException {
        try (Connection conn = HBaseConnection.getConnection();
             Table table = conn.getTable(TableName.valueOf(tableName))) {

            Scan scan = new Scan();
            ResultScanner scanner = table.getScanner(scan);

            for (Result result : scanner) {
                String rowKey = Bytes.toString(result.getRow());
                System.out.println("行键: " + rowKey);

                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);
                }
            }
        }
    }
}

import org.apache.hadoop.hbase.filter.*;

public class FilterOperations {

    // 列值过滤器
    public void filterByValue(String tableName, String columnFamily,
                             String column, String value) throws IOException {
        try (Connection conn = HBaseConnection.getConnection();
             Table table = conn.getTable(TableName.valueOf(tableName))) {

            Filter filter = new SingleColumnValueFilter(
                Bytes.toBytes(columnFamily),
                Bytes.toBytes(column),
                CompareOperator.EQUAL,
                Bytes.toBytes(value)
            );

            Scan scan = new Scan();
            scan.setFilter(filter);

            ResultScanner scanner = table.getScanner(scan);
            for (Result result : scanner) {
                System.out.println("匹配的行: " + Bytes.toString(result.getRow()));
            }
        }
    }

    // 前缀过滤器
    public void filterByPrefix(String tableName, String prefix)
            throws IOException {
        try (Connection conn = HBaseConnection.getConnection();
             Table table = conn.getTable(TableName.valueOf(tableName))) {

            Filter filter = new PrefixFilter(Bytes.toBytes(prefix));

            Scan scan = new Scan();
            scan.setFilter(filter);

            ResultScanner scanner = table.getScanner(scan);
            for (Result result : scanner) {
                System.out.println("前缀匹配: " + Bytes.toString(result.getRow()));
            }
        }
    }
}

// 1. 哈希前缀模式（解决热点问题）
String originalKey = "user123";
String hashedKey = String.valueOf(originalKey.hashCode() % 100) + "_" + originalKey;
// 结果: "23_user123"

// 2. 时间戳反转（最新数据在前）
long timestamp = System.currentTimeMillis();
String reversedTimestamp = String.valueOf(Long.MAX_VALUE - timestamp);
// 结果: 时间戳反转，最新数据排在前面

// 3. 复合键模式
String compositeKey = regionId + "_" + userId + "_" + timestamp;
// 结果: "region001_user123_1640995200000"

// 好的列族设计
TableDescriptor table = TableDescriptorBuilder
    .newBuilder(TableName.valueOf("user_profile"))
    .setColumnFamily(ColumnFamilyDescriptorBuilder
        .newBuilder(Bytes.toBytes("basic_info"))
        .setMaxVersions(1)
        .setCompressionType(Compression.Algorithm.SNAPPY)
        .build())
    .setColumnFamily(ColumnFamilyDescriptorBuilder
        .newBuilder(Bytes.toBytes("contact_info"))
        .setMaxVersions(3)
        .setTimeToLive(86400 * 30) // 30天
        .build())
    .build();

// 启用压缩
ColumnFamilyDescriptorBuilder cfBuilder =
    ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("cf1"));

// 使用 Snappy 压缩（平衡压缩率和速度）
cfBuilder.setCompressionType(Compression.Algorithm.SNAPPY);

// 使用 LZ4 压缩（速度优先）
cfBuilder.setCompressionType(Compression.Algorithm.LZ4);

// 使用 GZ 压缩（压缩率优先）
cfBuilder.setCompressionType(Compression.Algorithm.GZ);

// 创建用户行为表
public void createUserBehaviorTable() throws IOException {
    TableDescriptor table = TableDescriptorBuilder
        .newBuilder(TableName.valueOf("user_behavior"))
        .setColumnFamily(ColumnFamilyDescriptorBuilder
            .newBuilder(Bytes.toBytes("action"))
            .setMaxVersions(10)
            .setTimeToLive(86400 * 7) // 7天
            .build())
        .setColumnFamily(ColumnFamilyDescriptorBuilder
            .newBuilder(Bytes.toBytes("product"))
            .setMaxVersions(5)
            .build())
        .build();

    admin.createTable(table);
}

// 插入用户行为数据
public void recordUserAction(String userId, String action,
                           String productId, String timestamp)
        throws IOException {
    String rowKey = userId + "_" + timestamp;

    Put put = new Put(Bytes.toBytes(rowKey));
    put.addColumn(Bytes.toBytes("action"),
                 Bytes.toBytes("type"),
                 Bytes.toBytes(action));
    put.addColumn(Bytes.toBytes("product"),
                 Bytes.toBytes("id"),
                 Bytes.toBytes(productId));

    table.put(put);
}

// 获取用户最近行为
public List<String> getUserRecentActions(String userId, int limit)
        throws IOException {
    String startRow = userId + "_";
    String stopRow = userId + "_~";

    Scan scan = new Scan();
    scan.setStartRow(Bytes.toBytes(startRow));
    scan.setStopRow(Bytes.toBytes(stopRow));
    scan.setReversed(true); // 倒序，获取最新数据
    scan.setMaxResultSize(limit);

    List<String> actions = new ArrayList<>();
    ResultScanner scanner = table.getScanner(scan);

    for (Result result : scanner) {
        String action = Bytes.toString(result.getValue(
            Bytes.toBytes("action"), Bytes.toBytes("type")));
        actions.add(action);
    }

    return actions;
}

// 创建日志表
public void createLogTable() throws IOException {
    TableDescriptor table = TableDescriptorBuilder
        .newBuilder(TableName.valueOf("app_logs"))
        .setColumnFamily(ColumnFamilyDescriptorBuilder
            .newBuilder(Bytes.toBytes("log"))
            .setMaxVersions(1)
            .setCompressionType(Compression.Algorithm.SNAPPY)
            .build())
        .setColumnFamily(ColumnFamilyDescriptorBuilder
            .newBuilder(Bytes.toBytes("meta"))
            .setMaxVersions(1)
            .build())
        .build();

    admin.createTable(table);
}

// 按时间范围查询日志
public void queryLogsByTimeRange(String startTime, String endTime)
        throws IOException {
    Scan scan = new Scan();
    scan.setStartRow(Bytes.toBytes(startTime));
    scan.setStopRow(Bytes.toBytes(endTime));

    ResultScanner scanner = table.getScanner(scan);
    for (Result result : scanner) {
        String logContent = Bytes.toString(result.getValue(
            Bytes.toBytes("log"), Bytes.toBytes("content")));
        System.out.println(logContent);
    }
}