簡簡單單用一下 Hbase

一、Hbase 介紹

//hbase.apache.org/book.html#_preface

//blogs.apache.org/hbase/

//research.google.com/archive/bigtable.html

什麼是Hbase？

hadoop 資料庫：分散式、可伸縮、大數據存儲。

二、Hbase client

最開始引入 hbase-client，服務有使用【google/protobuf/wrappers.proto】，有很多包衝突，所以直接使用了 habase-shade-client： 

<dependency>
  <groupId>org.apache.hbase</groupId>
  <artifactId>hbase-shaded-client</artifactId>
  <version>${hbase.shade.client.version}</version>
</dependency>

三、Hbase 配置

  • hbase.zookeeper.quorum
    zookeeper server 地址，逗號分割。本地模式和偽集群模式下，默認為 127.0.0.1

  • hbase.zookeeper.property.clientPort
    zookeeper server 埠，默認 2181

  • hbase.client.retries.number
    hbase client 所有操作的重試上限，默認 15。client 首先等待 hbase.client.pause 執行第一次重試，之後每隔 10s 再次執行。

  • hbase.rpc.timeout
    hbase client 一次 rpc 操作的超時時間（超時基於ping檢查），默認60000ms，觸發則拋出 TimeoutException 異常。

  • hbase.client.operation.timeout
    hbase client 一次操作的總的時間限制， 默認 1200000ms，觸發則直接拋出 SocketTimeoutException 異常。

  • 示例：
    @Configuration
public class HBaseConfig {
    @Value("${hbase.zookeeper.quorum}")
    private String hbaseZkQuorum;
    @Value("${hbase.zookeeper.property.clientPort:2181}")
    private String hbaseZkPort;
    @Value("${hbase.client.retries.number:2}")
    private String hbaseClientRetry;
    @Value("${hbase.rpc.timeout:2000}")
    private String hbaseRpcTimeout;
    @Value("${hbase.client.operation.timeout:3000}")
    private String hbaseClientOperationTimeout;
    @Bean
    public Connection hbaseConnection() throws IOException {
        org.apache.hadoop.conf.Configuration hbaseConfig = HBaseConfiguration.create();
        hbaseConfig.set("hbase.zookeeper.property.clientPort", hbaseZkPort);
        hbaseConfig.set("hbase.zookeeper.quorum", hbaseZkQuorum);
        hbaseConfig.set("hbase.client.retries.number", hbaseClientRetry);
        hbaseConfig.set("hbase.client.operation.timeout", hbaseClientOperationTimeout);
        hbaseConfig.set("hbase.rpc.timeout", hbaseRpcTimeout);
        return ConnectionFactory.createConnection(hbaseConfig);
    }
    @Bean
    public HbaseSimpleTemplate hbaseSimpleTemplate(@Qualifier("hbaseConnection") Connection hbaseConnection) {
        return new HbaseSimpleTemplate(hbaseConnection);
    }
}

四、關於 Connection

1、Connection 是什麼？

集群 connection 封裝了底層和實際 hbase server 及 zookeeper 的連接。由 ConnectionFactory 創建並由發起端維護其整個生命周期。

承載了服務發現（hbase master 及 region server）及本地快取維護（存儲及更新）邏輯。所以基於此鏈接實例化而來的 Table 和 Admin 共享此資訊。

2、Connection 怎麼使用？

Connection 創建是一個很重的操作。

Connection 實現是 thread-safe 的。

所以通常的操作時，一次創建，到處使用。

這裡我們通過 @Bean 註解，將 connection 實例交由 spring 管理，維護其從創建，使用到銷毀的整個生命周期。

三、HbaseSimpleTemplate

Hbase Connection 數據操作封裝：

row->column->all cells

row->column->cells

rows->column->cells

public class HbaseSimpleTemplate {
    private Connection hbaseConnection;
    public HbaseSimpleTemplate(Connection hbaseConnection) {
        this.hbaseConnection = hbaseConnection;
    }
    /**
     * 結果映射map
     *
     * @param result
     * @return
     */
    private Map<String, String> resultToMap(Result result) {
        if (result == null || result.isEmpty()) {
            return new HashMap<>();
        }
        return result.listCells().stream().collect(
                Collectors.toMap(cell -> Bytes.toString(CellUtil.cloneQualifier(cell)), cell -> Bytes.toString(CellUtil.cloneValue(cell))));
    }
    /**
     * 查詢
     * @param tableName
     * @param rowName
     * @param familyName
     * @return
     * @throws IOException
     */
    public Map<String, String> get(String tableName, String rowName, String familyName) throws IOException {
        Map<String, Map<String, String>> resultMap = get(tableName, Collections.singletonList(rowName), familyName, null);
        return resultMap.values().stream().findFirst().orElse(new HashMap<>());
    }
    /**
     *
     * @param tableName
     * @param rowName
     * @param familyName
     * @param qualifiers
     * @return
     * @throws IOException
     */
    public Map<String, String> get(String tableName, String rowName, String familyName, List<String> qualifiers) throws IOException {
        Map<String, Map<String, String>> resultMap = get(tableName, Collections.singletonList(rowName), familyName, qualifiers);
        return resultMap.values().stream().findFirst().orElse(new HashMap<>());
    }
    /**
     * 批量查詢
     *
     * @param tableName
     * @param rowNames
     * @param familyName
     * @return
     * @throws IOException
     */
    public Map<String, Map<String, String>> get(String tableName, List<String> rowNames, String familyName, List<String> qualifiers) throws IOException {
        Map<String, Map<String, String>> resultMap = new HashMap<>();
        List<Get> gets = new ArrayList<>();
        rowNames.forEach(rowName -> {
            Get get = new Get(rowName.getBytes());
            if (CollectionUtils.isNotEmpty(qualifiers)) {
                qualifiers.forEach(qualifier -> get.addColumn(familyName.getBytes(), qualifier.getBytes()));
            } else {
                get.addFamily(familyName.getBytes());
            }
            gets.add(get);
        });
        Arrays.stream(hbaseConnection.getTable(TableName.valueOf(tableName)).get(gets))
                .forEach(result -> {
                    Map<String, String> kvMap = resultToMap(result);
                    String id = MapUtils.getString(kvMap, "id");
                    if (StringUtils.isNotBlank(id)) {
                        resultMap.put(id, kvMap);
                    }
                });
        return resultMap;
    }
    /**
     * 寫入 qualifier
     *
     * @param tableName
     * @param rowName
     * @param familyName
     * @param qualifier
     * @param value
     * @return
     * @throws IOException
     */
    public boolean put(String tableName, String rowName, String familyName, String qualifier, String value) throws IOException {
        Map<String, String> qv = new HashMap<>();
        qv.put(qualifier, value);
        put(tableName, rowName, familyName, qv);
        return true;
    }
    /**
     * 寫入 qualifiers
     *
     * @param tableName
     * @param rowName
     * @param familyName
     * @param qualifierValues
     * @return
     * @throws IOException
     */
    public boolean put(String tableName, String rowName, String familyName, Map<String, String> qualifierValues) throws IOException {
        if (MapUtils.isEmpty(qualifierValues)) {
            return false;
        }
        List<Put> puts = new ArrayList<>();
        qualifierValues.forEach((qualifier, value) -> puts.add(new Put(rowName.getBytes()).addColumn(familyName.getBytes(), qualifier.getBytes(), value.getBytes())));
        hbaseConnection.getTable(TableName.valueOf(tableName)).put(puts);
        return true;
    }
    /**
     * 刪除 
     *
     * @param tableName
     * @param rowName
     * @param familyName
     * @return
     * @throws IOException
     */
    public boolean del(String tableName, String rowName, String familyName) throws IOException {
        Delete delete = new Delete(rowName.getBytes());
        delete.addFamily(familyName.getBytes());
        hbaseConnection.getTable(TableName.valueOf(tableName)).delete(delete);
        return true;
    }
    /**
     * 刪除 qualifier
     *
     * @param tableName
     * @param rowName
     * @param familyName
     * @param qualifiers
     * @return
     * @throws IOException
     */
    public boolean delQualifiers(String tableName, String rowName, String familyName, List<String> qualifiers) throws IOException {
        Delete delete = new Delete(rowName.getBytes());
        qualifiers.forEach(qualifier -> delete.addColumn(familyName.getBytes(), qualifier.getBytes()));
        hbaseConnection.getTable(TableName.valueOf(tableName)).delete(delete);
        return true;
    }
}

getTable：

獲取 Table 實現用以訪問表數據。

Table 非 thread-safe 的並且其創建很輕量，所以執行緒內使用需要單獨創建（不需要且不應該快取和池化）。 

 

Tags:

VirMach 便宜 VPS