package com.zoujc.Utils;

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


/**
 * javaAPI操作HBase     get put namespace scan filter cache
 */
 

public class HBaseTest {
   static Configuration conf = null;
   static {
      conf = HBaseConfiguration.create();
      conf.set("hbase.zookeeper.quorum", "node1:2181,node2:2181,node3:2181");
      conf.set("hbase.rootdir","hdfs://hadoopcluster/hbase");
   }

   public static void main(String[] args)
 
{
      /**createNamespace("ns1");        
       * createTable("apitable","info");        
       * addColumnFamily("apitable","secret");        
       * putData("apitable","1","info","name","sid");        
       * putData("apitable","1","info","age","27");        
       * putData("apitable","1","secret","idcart","123");        
       * putData("apitable","2","info","age","28");        
       * putData("apitable","2","info","name","zhangsi");        
       * putData("apitable","3","info","name","lisi");        
       * putData("apitable","3","info","school","cqut");        
       * putData("apitable","4","info","name","wangwu");        
       * getData("apitable","1","info","name");*/
 
         
      //scanData("apitable","1","3","info");         
      // filter("apitable","1");
   }
   /**
    * 建立namespace
    */
   public static void createNamespace(String namespace) throws Exception{
      HBaseAdmin admin = new HBaseAdmin(conf);

      //建立namespace
      NamespaceDescriptor descriptor = NamespaceDescriptor.create(namespace).build();
      admin.createNamespace(descriptor);

      admin.close();
   }

   /**
    * put
    */
   public static void putData(String tableName, String rowKey, String columnFamily, String column, String value) throws Exception{
      HTable table = getTable(tableName);
      Put put = new Put(Bytes.toBytes(rowKey));
      put.add(Bytes.toBytes(columnFamily),Bytes.toBytes(column),Bytes.toBytes(value));
      table.put(put);

      table.close();
   }

   /**
    * get
    */
   public static void getData(String tableName,String rowKey,String columnFamily,String column) throws Exception{
      HTable table = getTable(tableName);

      //一定要知道rowKey
      Get get = new Get(Bytes.toBytes(rowKey));
      Result result = table.get(get);

      //在知道列名和列族的情況下指定單元格
      byte[] value = result.getValue(Bytes.toBytes(columnFamily),Bytes.toBytes(column));
      System.out.println(new String(value));

      //在只知道rowkey的情況下，遍歷每個單元格
      printResult(result);
      //System.out.println(new String(cell.getFamily())+":"+new String(cell.getQualifier())+":"+new String(cell.getValue()));
   }

   /**
    * scan
    * 一個Result是一行資料
    * ResultScanner是多行資料
    * */
   public static void scanData(String tableName,String startRowKey,String stopRowKey,String columnFamily){
      HTable table = null;
      try {
         table = getTable(tableName);
         //全表掃描
         Scan scan = new Scan();

         //根據rowkey範圍查詢 開始是包括這個rowkey  結束是不包括這個rowkey
         scan.setStartRow(Bytes.toBytes(startRowKey));
         scan.setStopRow(Bytes.toBytes(stopRowKey));

         //查詢指定列族下的
         scan.addFamily(Bytes.toBytes(columnFamily));

         //是否本地快取
         scan.setCacheBlocks(false);   //預設是true
         //面向列的快取，每次返回3列
         scan.setBatch(3);
         //面向行的快取 每次返回的行有2行 如果一共有10行資料，那麼就會讀取5次
         scan.setCaching(2);

         //許可權設定
         //scan.setACL();

         ResultScanner results = table.getScanner(scan);
         for(Result rs: results){
            printResult(rs);
         }
      } catch (Exception e) {
         e.printStackTrace();
      } finally {
         if(table != null){
            IOUtils.closeStream(table);
         }
      }
   }

   //過濾器
   public static void filter(String tableName,String prefix){
      HTable table = null;

      try {
         table = getTable(tableName);
         Scan scan = new Scan();

         Filter filter = null;

         //根據列的字首查詢 不是列族！
//    filter = new ColumnPrefixFilter(Bytes.toBytes(prefix));
         //根據rowkey的字首查詢！
//    filter = new PrefixFilter(Bytes.toBytes(prefix));
         //分頁過濾，查詢3條
//    filter = new PageFilter(3);

         //值比較
         ByteArrayComparable comp = null;
         //列族info的name的值為lisi 就查詢出來
         comp = new SubstringComparator("lisi");
         filter = new SingleColumnValueExcludeFilter(Bytes.toBytes("info"),Bytes.toBytes("name"),CompareFilter.CompareOp.EQUAL,comp);
         scan.setFilter(filter);

         ResultScanner results = table.getScanner(scan);
         for(Result rs: results){
            printResult(rs);
         }
      } catch (Exception e) {
         e.printStackTrace();
      } finally {
         if(table != null){
            IOUtils.closeStream(table);
         }
      }
   }

   /**
    * 打印出一行資料的每個單元格
    */
   public static void printResult(Result rs){
      for(Cell cell: rs.rawCells()){
         System.out.println("row key :" + Bytes.toString(CellUtil.cloneRow(cell)));
         System.out.println("column family :" + Bytes.toString(CellUtil.cloneFamily(cell)));
         System.out.println("column name :" + Bytes.toString(CellUtil.cloneQualifier(cell)));
         System.out.println("value :" + Bytes.toString(CellUtil.cloneValue(cell)));
      }
   }

   /**
    * 獲取表
    */
   public static HTable getTable(String tableName) throws Exception{
      return new HTable(conf,Bytes.toBytes(tableName));
   }

   /**
    * 建立表
    */
   public static void createTable(String tableName,String columnFamily) throws Exception{
      //hbase表結果操作的入口
      HBaseAdmin admin = new HBaseAdmin(conf);
      //判斷表是否存在
      boolean b = admin.tableExists(tableName);
      if(b){
         //禁用表
         admin.disableTable(tableName);
         //刪除表
         admin.deleteTable(tableName);
      }
      //設定表名
      HTableDescriptor table = new HTableDescriptor(TableName.valueOf(tableName));
      //設定列族
      table.addFamily(new HColumnDescriptor(Bytes.toBytes(columnFamily)));
      //建立表
      admin.createTable(table);

      admin.close();
   }

   /**
    * 給已有的表新增列族
    */
   public static void addColumnFamily(String tableName,String columnFamily) throws Exception{
      //hbase表結果操作的入口
      HBaseAdmin admin = new HBaseAdmin(conf);
      //判斷表是否存在
      boolean b = admin.tableExists(tableName);
      if(b){
         //新增列族
         admin.addColumn(tableName,new HColumnDescriptor(Bytes.toBytes(columnFamily)));
      }
   }
}