package com.dajiangtai.hadoop.hbase;

import java.io.IOException;

import javax.ws.rs.PUT;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.hbase.HBaseConfiguration;

import org.apache.hadoop.hbase.HColumnDescriptor;

import org.apache.hadoop.hbase.HTableDescriptor;

import org.apache.hadoop.hbase.KeyValue;

import org.apache.hadoop.hbase.MasterNotRunningException;

import org.apache.hadoop.hbase.ZooKeeperConnectionException;

import org.apache.hadoop.hbase.client.Delete;

import org.apache.hadoop.hbase.client.Get;

import org.apache.hadoop.hbase.client.HBaseAdmin;

import org.apache.hadoop.hbase.client.HTable;

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

import org.apache.hadoop.hbase.client.Result;

import org.apache.hadoop.hbase.client.ResultScanner;

import org.apache.hadoop.hbase.client.Scan;

import org.apache.hadoop.hbase.util.Bytes;

import com.sun.xml.internal.ws.api.ha.StickyFeature;

public class HbaseTest {

   public static Configuration conf;

   static {

       conf = HBaseConfiguration.create();//第一步

       conf.set("hbase.zookeeper.quorum", "dajiangtai1,dajiangtai2,dajiangtai3,dajiangtai4,dajiangtai5");

       conf.set("hbase.zookeeper.property.clientPort", "");

       conf.set("hbase.master", "dajiangtai1:60000");

   }

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

       //createTable("member");

      //insertDataByPut("member");

       QueryByGet("member");

       //QueryByScan("member");

      //deletData("member");

   }

   public static void createTable(String tableName) throws MasterNotRunningException,ZooKeeperConnectionException,

   IOException{

       HBaseAdmin hBaseAdmin = new HBaseAdmin(conf);//创建HBaseAdmin对象

      // 先判断这个表是否存在
        if(hBaseAdmin.tableExists(tableName)){

           hBaseAdmin.disableTable(tableName);

           hBaseAdmin.deleteTable(tableName);

       }

      HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);//将表名称传进去实例化HTableDescriptor这个表

      //添加family

      tableDescriptor.addFamily(new HColumnDescriptor("address"));

      tableDescriptor.addFamily(new HColumnDescriptor("info"));

      hBaseAdmin.createTable(tableDescriptor);//创建表

      hBaseAdmin.close();//释放资源

   }

  //插入数据方法

   public static void insertDataByPut(String tableName) throws IOException

   {

       HTable table = new HTable(conf,tableName);

       Put put1= new Put(getBytes("djt"));

       put1.add(getBytes("address"),getBytes("country"),getBytes("china"));

       put1.add(getBytes("address"),getBytes("province"),getBytes("beijing"));

       put1.add(getBytes("address"),getBytes("city"),getBytes("beijing"));

       put1.add(getBytes("info"),getBytes("age"),getBytes(""));

       put1.add(getBytes("info"),getBytes("birthday"),getBytes("1988-12-12"));

       put1.add(getBytes("info"),getBytes("company"),getBytes("dajiangtai"));

       table.put(put1); //插入数据

       table.close();  //释放资源

   }

//查询数据

   public static void QueryByGet(String tableName) throws IOException{

       HTable table =new HTable(conf,tableName);

       Get get = new Get(getBytes("djt")); //根据rowkey查询

       Result r=table.get(get); //执行操作

       System.out.println("获得到rowkey:"+new String(r.getRow()));

    for(KeyValue keyValue : r.raw()){

        System.out.println("列簇："  +  new String(keyValue.getFamily())

        + "====列" + new String(keyValue.getQualifier()) + "====值"

        +  new String(keyValue.getValue())  );

    }

    table.close();

   }

   public static void QueryByScan(String tableName) throws IOException{

       HTable table =new HTable(conf,tableName);

       Scan scan = new Scan();

       scan.addColumn(getBytes("info"),getBytes("company"));

       ResultScanner scanner = table.getScanner(scan);

       for(Result r : scanner) {

           System.out.println("获得到rowkey:" +  new String(r.getRow()));

           for(KeyValue kv : r.raw()){

               System.out.println("列簇："  + new String(kv.getFamily())

        + "====列" + new String(kv.getQualifier()) + "====值"

        +  new String(kv.getValue())  );

           }

       }

       scanner.close();

       table.close();

   }

   public static void deleteData(String tableName) throws IOException{

       HTable table = new HTable(conf,tableName);

       Delete delete = new Delete(getBytes("djt"));

       delete.deleteColumn(getBytes("info"),getBytes("age"));

       table.delete(delete);

       table.close();

   }

   //转化byte数组

   public static byte[] getBytes(String str){

       if(str == null)

           str="";

       return Bytes.toBytes(str);

   }

}