package com.zoujc.Utils;

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

import java.io.IOException;
import java.
 
util.HashMap;
import java.util.Map;


/**
 * Hbase操作工具類
 * java工具類建議採用單例模式封裝
 */
public class HbaseUtils {

   HBaseAdmin admin = null;
   Configuration conf = null;

   //私有構造方法，載入重要配置
   private HbaseUtils(){
      conf = new Configuration();
      //zookeeper叢集的URL配置，多個host中間用逗號（,）分割
      conf.set("hbase.zookeeper.quorum"
 
,"node1:2181,node2:2181,node3:2181");
      //HBase叢集中所有RegionServer共享目錄，用來持久化HBase的資料，一般設定的是hdfs的檔案目錄，如hdfs://namenode.example.org:9000/hbase
      conf.set("hbase.rootdir","hdfs://hadoopcluster/hbase");

      try {
         admin = new HBaseAdmin(conf);
      } catch (IOException e) {
         e.printStackTrace
 
();
      }
   }

   //getInstance這個方法在單例模式用的甚多，為了避免對記憶體造成浪費，直到需要例項化該類的時候才將其例項化，所以用getInstance來獲取該物件，
   //至於其他時候，也就是為了簡便而已，為了不讓程式在例項化物件的時候，不用每次都用new關鍵字，索性提供一個instance方法，不必一執行這個類就
   //初始化，這樣做到不浪費系統資源！單例模式 可以防止 資料的衝突，節省記憶體空間
   private static HbaseUtils instance = null;

   //單例模式例項化物件
   public static synchronized HbaseUtils getInstance(){
      if(null == instance){
         instance = new HbaseUtils();
      }
      return instance;
   }

   /**
    * 根據表名獲取到HTable例項
    */
   public HTable getTable(String tableName){

      HTable table = null;

      try {
         table = new HTable(conf,tableName);
      } catch (IOException e) {
         e.printStackTrace();
      }

      return table;
   }

   /**
    * 新增一條記錄到Hbase表
    * @param tableName    Hbase表名
    * @param rowKey      Hbase表的rowkey
    * @param cf         Hbase表的列族columnfamily
    * @param cloumn      Hbase表的列
    * @param value       寫入Hbase表的值
    */
   public void put(String tableName,String rowKey,String cf,String cloumn,String value){

      HTable table = getTable(tableName);

      Put put = new Put(Bytes.toBytes(rowKey));
      put.add(Bytes.toBytes(cf),Bytes.toBytes(cloumn),Bytes.toBytes(value));

      try {
         table.put(put);
      } catch (IOException e) {
         e.printStackTrace();
      }
   }

   //根據輸入表名和條件獲取表中記錄數
   public Map<String,Long> query(String tableName,String condition) throws Exception{

      Map<String,Long> map = new HashMap<String, Long>();

      HTable table = getTable(tableName);
      String cf = "info";
      String qualifier = "click_count";

      Scan scan = new Scan();

      //根據rowKey的字首來查詢
      Filter filter = new PrefixFilter(Bytes.toBytes(condition));
      scan.setFilter(filter);

      ResultScanner rs = table.getScanner(scan);
      for(Result result: rs){
         String row = Bytes.toString(result.getRow());
         Long clickCount = Bytes.toLong(result.getValue(cf.getBytes(),qualifier.getBytes()));
         map.put(row,clickCount);
      }

      return map;
   }

   public static void main(String[] args) throws Exception {

//    test連線
//    HTable table = HbaseUtils.getInstance().getTable("HbaseTableName");
//    System.out.println(table.getName().getNameAsString());

      String tablename = "HbaseTableName";
      String rowKey = "HbaseTableRowKey";
      String cf = "HbaseTableCF";
      String cloumn = "HbaseTableCloumn";
      String value = "HbaseTableValue";

      HbaseUtils.getInstance().put(tablename,rowKey,cf,cloumn,value);

      Map<String,Long> map = HbaseUtils.getInstance().query("HbaseTableName","2018-10-17");

      for(Map.Entry<String,Long> entry: map.entrySet()){
         System.out.println(entry.getKey() + ":" + entry.getValue());
      }
   }
}