package testHBase;

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.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;

public class TestHBase {

	static Configuration cfg = HBaseConfiguration.create();  //得到配置对象

	public static void create(String tableName ,String columnFamily) throws Exception{

		HBaseAdmin admin = new HBaseAdmin(cfg);  //得到HBaseAdmin对象，用于操作HBase

		if(admin.tableExists(tableName)){

			System.out.println("Table" + tableName + " Exists!");

			System.exit(0);

		}

		else{

			HTableDescriptor tableDesc = new HTableDescriptor(tableName); //得到表描述对象  创建表时用到

			tableDesc.addFamily(new HColumnDescriptor(columnFamily)); //HColumnDescriptor 列族描述对象

			admin.createTable(tableDesc);  //创建表

			System.out.println("Table " + tableName + " Success!");

		}

	}

	static void put(String tableName,String row,String columnFamily,String column,String data) throws Exception{

		HTable table = new HTable(cfg,tableName);  //得到表对象

		Put put = new Put(Bytes.toBytes(row));		//插入数据

		put.add(Bytes.toBytes(columnFamily), Bytes.toBytes(column),Bytes.toBytes(data));

		table.put(put);

		System.out.println("put " + row +"," + "columnFamily:" + column + "," + data);

	}

	static void get(String tableName,String row) throws Exception{

		HTable table = new HTable(cfg,tableName);

		Get get = new Get(Bytes.toBytes(row));//获取行数据

		Result result = table.get(get);

		System.out.println("Get: " + result);

	}

	static void scan(String tableName) throws Exception{

		HTable table = new HTable(cfg,tableName);

		Scan scan = new Scan(Bytes.toBytes(tableName)); ///扫描全表

		ResultScanner rs = table.getScanner(scan);

		for(Result r : rs){

			System.out.println("Scan: " + r);

		}

	}

	static boolean delete(String tableName) throws Exception{

		HBaseAdmin admin = new HBaseAdmin(cfg);

		if(admin.tableExists(tableName)){

			admin.disableTable(tableName);  //先disable

			System.out.println("Disable Table " + tableName + " Success!");

			admin.deleteTable(tableName); //再删除

			System.out.println("Delete Table " + tableName + " Success!");

		}

		return true;

	}

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

		cfg.set("hbase.master", "hadoop:60000");

		cfg.set("hbase.rootdir", "hdfs://hadoop:9000/hbase");

		cfg.set("hbase.zookeeper.quorum", "hadoop");

		String tableName = "hbase_test";

		String columnFamily = "cf";

		create(tableName, columnFamily);

		put(tableName,"row1",columnFamily,"c1","data");

		scan(tableName);

		get(tableName,"row1");

		delete(tableName);

	}

}