[大数据]大数据入门--HBase（二）常用API案例

HBase常用API案例

概述

抛砖引玉篇，不废话。主要演示了HBase常见操作API

• DDL
  • namespace 创建 删除
  • table 创建 删除
• DML
  • 数据的增改（改==增，都是put操作）、删、查

话不多说，撸代码！！！

package com.xbz.study.bigdata.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

/**
* 封装Hbase API的常用操作
* DDL：
* 创建名称空间
* 创建表
* 判断表是否存在
* 删除表
* DML：
* 创建数据
* 更新数据
* 删除数据
* 查询数据
*/
public class ApiUtil {
   private static final Configuration configuration;

   static {
       configuration = HBaseConfiguration.create();
       configuration.set("hbase.zookeeper.quorum", "hadoop101:2181,hadoop102:2181,hadoop103:2181");
   }

   private static Connection openConnection() throws IOException {
       return ConnectionFactory.createConnection(configuration);
   }

   /**
    * 创建命名空间
    *
    * @param nameSpace
    */
   public static boolean createNameSpace(String nameSpace) throws IOException {
       Connection connection = openConnection();
       Admin admin = null;
       try {
           admin = connection.getAdmin();
           NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(nameSpace).build();
           admin.createNamespace(namespaceDescriptor);

       } catch (NamespaceExistException ex) {
           return false;
       } catch (Exception ex) {
           ex.printStackTrace();
           return false;
       } finally {
           if (admin != null) {
               admin.close();
           }
           connection.close();
       }
       return true;
   }

   /**
    * 删除命名空间
    *
    * @param nameSpace
    * @return
    * @throws IOException
    */
   public static boolean deleteNameSpace(String nameSpace) throws IOException {
       Connection connection = openConnection();
       Admin admin = null;
       try {
           admin = connection.getAdmin();
           admin.deleteNamespace(nameSpace);
       } catch (NamespaceNotFoundException ex) {
           return false;
       } catch (Exception ex) {
           ex.printStackTrace();
           return false;
       } finally {
           if (admin != null) {
               admin.close();
           }
           connection.close();
       }
       return true;
   }

   /**
    * 创建表
    *
    * @param tableName
    * @param cfs
    */
   public static boolean createTable(String tableName, String... cfs) throws IOException {
       if (cfs.length == 0) {
           throw new IllegalArgumentException("列族不能为空");
       }
       Connection connection = openConnection();
       Admin admin = null;
       try {
           admin = connection.getAdmin();
           boolean exists = admin.tableExists(TableName.valueOf(tableName));
           if (exists) {
               return false;
           }
           HTableDescriptor hTableDesc = new HTableDescriptor(TableName.valueOf(tableName));
           for (String cf : cfs) {
               hTableDesc.addFamily(new HColumnDescriptor(cf));
           }
           admin.createTable(hTableDesc);
       } catch (Exception ex) {
           ex.printStackTrace();
           return false;
       } finally {
           admin.close();
           connection.close();
       }
       return true;
   }

   /**
    * 删除表
    *
    * @param tableName
    * @return
    * @throws IOException
    */
   public static boolean deleteTable(String tableName) throws IOException {
       Connection connection = openConnection();
       Admin admin = null;
       try {
           admin = connection.getAdmin();
           //判断表是否存在
           TableName tName = TableName.valueOf(tableName);
           boolean exists = admin.tableExists(tName);
           if (!exists) {
               return true;
           }
           //如果存在，判断表是否已禁用
           boolean tableDisabled = admin.isTableDisabled(tName);
           if (!tableDisabled) {
               admin.disableTable(tName);
           }
           admin.deleteTable(tName);
       } catch (Exception ex) {
           ex.printStackTrace();
           return false;
       } finally {
           admin.close();
           connection.close();
       }
       return true;
   }

   /**
    * 添加数据
    *
    * @param tableName
    * @param cf        列族
    * @param cn        列名
    * @param value     值
    */
   public static void put(String tableName, String rowKey, String cf, String cn, String value) throws IOException {
       Connection connection = openConnection();
       Table table = null;
       try {
           table = connection.getTable(TableName.valueOf(tableName));
           Put put = new Put(Bytes.toBytes(rowKey));
           put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(cn), Bytes.toBytes(value));
           table.put(put);
       } catch (Exception ex) {
           throw new RuntimeException(ex);
       } finally {
           if (table != null) {
               table.close();
           }
           connection.close();
       }
   }

   /**
    * 删除一行或者多行数据
    *
    * @param tableName
    * @param rowKeys
    */
   public static void deleteRow(String tableName, String... rowKeys) throws IOException {
       if (rowKeys == null || rowKeys.length == 0) {
           throw new IllegalArgumentException("至少指定一个rowKey");
       }
       Connection connection = openConnection();
       Table table = null;
       try {
           table = connection.getTable(TableName.valueOf(tableName));
           List<Delete> deletes = Arrays.stream(rowKeys).map((i) -> new Delete(Bytes.toBytes(i))).collect(Collectors.toList());
           table.delete(deletes);
       } catch (Exception ex) {
           throw new RuntimeException(ex);
       } finally {
           if (table != null) {
               table.close();
           }
           connection.close();
       }
   }

   /**
    * 删除指定列
    *
    * @param tableName
    * @param cf        列族
    * @param cn        列名称
    */
   public static void deleteCol(String tableName, String rowKey, String cf, String cn) throws IOException {
       Connection connection = openConnection();
       Table table = null;
       try {
           table = connection.getTable(TableName.valueOf(tableName));
           Delete delete = new Delete(Bytes.toBytes(rowKey));
           /*
            * 注意使用addColumns是删除多个版本
            * addCloumn是删除指定版本，如果此时有多个版本数据，则上个版本数据会冒出来，慎用！！！
            */
           delete.addColumns(Bytes.toBytes(cf), Bytes.toBytes(cn));
           table.delete(delete);
       } catch (Exception ex) {
           throw new RuntimeException(ex);
       } finally {
           if (table != null) {
               table.close();
           }
           connection.close();
       }
   }

   /**
    * 获取列数据
    *
    * @param tableName
    * @param rowKey
    * @param cf
    * @param cn
    * @return
    * @throws IOException
    */
   public static String getCol(String tableName, String rowKey, String cf, String cn) throws IOException {
       Connection connection = openConnection();
       Table table = null;
       try {
           table = connection.getTable(TableName.valueOf(tableName));
           Get get = new Get(Bytes.toBytes(rowKey));
           get.addColumn(Bytes.toBytes(cf), Bytes.toBytes(cn));
           Result result = table.get(get);
           return Bytes.toString(result.value());
       } catch (Exception ex) {
           throw new RuntimeException(ex);
       } finally {
           if (table != null) {
               table.close();
           }
           connection.close();
       }
   }

   /**
    * 扫描多行
    *
    * @param tableName
    * @param beginRowKey
    * @param endRowKey
    * @throws IOException
    */
   public static void scan(String tableName, String beginRowKey, String endRowKey) throws IOException {
       Connection connection = openConnection();
       Table table = null;
       try {
           table = connection.getTable(TableName.valueOf(tableName));
           Scan scan = new Scan();
           scan.setStartRow(Bytes.toBytes(beginRowKey));
           scan.setStopRow(Bytes.toBytes(endRowKey));
           ResultScanner scanner = table.getScanner(scan);
           System.out.println("---------------------------------------------------");
           scanner.forEach((result -> {
               String row = Bytes.toString(result.getRow());
               System.out.println("row:" + row);
               Cell[] cells = result.rawCells();
               for (Cell cell : cells) {
//                    String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                   String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                   String cn = Bytes.toString(CellUtil.cloneQualifier(cell));
                   String value = Bytes.toString(CellUtil.cloneValue(cell));
                   String timestamp = cell.getTimestamp() + "";
                   System.out.println(cf + ":" + cn + "=" + value + "," + timestamp);
               }
               System.out.println("--------------------------------------------------------------");
           }));
       } catch (Exception ex) {
           throw new RuntimeException(ex);
       } finally {
           if (table != null) {
               table.close();
           }
           connection.close();
       }
   }


   public static void main(String[] args) throws IOException {
       //创建命名空间
       String nameSpace = "study02";
//        System.out.println("创建命名空间："+nameSpace+":"+(createNameSpace(nameSpace)?"成功":"失败"));
//        System.out.println("删除命名空间：" + nameSpace + ":" + (deleteNameSpace(nameSpace) ? "成功" : "失败"));
       String tableName = "tab01";
//        System.out.println("创建表："+createTable(tableName,"info1","info2"));
//        System.out.println("删除表："+deleteTable(tableName));
//        put(tableName,"001","info1","name","zhangsan");
//        deleteRow(tableName,"001","002");
//        deleteCol(tableName, "001", "info1", "name");
//        System.out.println(getCol(tableName, "001", "info1", "age"));
       scan(tableName, "001", "002");
   }
}