package chapter04;

import org.apache.commons.lang.StringUtils;

import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.hbase.Cell;

import org.apache.hadoop.hbase.CellUtil;

import org.apache.hadoop.hbase.HBaseConfiguration;

import org.apache.hadoop.hbase.TableName;

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

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

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

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

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

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

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

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

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

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

public class ExampleForHbase {

	//配置类

	public static Configuration configuration ;

	// 连接类

	public static Connection connection ;

	// 管理类

	public static Admin admin;

	public static void main(String[] args) {

		init();

		//创建表

		//createTable("Score",new String[]{"sname","course"});

		// 删除表

		//dropTable("Score");

		// 查询所有的表

		//listTable();

		// 新增一条数据

		//insertRow("Score", "98001", "sname", "", "张三");

		// insertRow("Score","98001","course","math","80");

		// 查询数据

		getData("Score", "98001", "course", "math");

		// 删除数据

		deleteRow("Score", "98001", "sname", "");

		getData("Score", "98001", "sname","");

		close();

	}

	/**

	 * 初始化

	 */

	public static void init() {

		configuration = HBaseConfiguration.create();

		configuration.set("hbase.rootdir", "hdfs://localhost:9000/hbase");

		try {

			connection = ConnectionFactory.createConnection(configuration);

			admin = connection.getAdmin();

		} catch (Exception e) {

			e.printStackTrace();

		}

	}

	/**

	 * 关闭

	 */

	public static void close() {

		try {

			if(admin!=null) {

				admin.close();

			}

			if(null != connection) {

				connection.close();

			}

		} catch (Exception e) {

			e.printStackTrace();

		}

	}

	/**

	 * 创建表

	 * @param tabName

	 * @param colFamily

	 */

	public static void createTable(String tabName,String[] colFamily) {

		try {

			TableName tableName = TableName.valueOf(tabName);

			if(admin.tableExists(tableName)) {

				System.out.println("Table is exist!");

			}else {

				TableDescriptorBuilder tableBuilder = TableDescriptorBuilder.newBuilder(tableName);

				for(String str:colFamily) {

					tableBuilder.setColumnFamily(ColumnFamilyDescriptorBuilder.of(str)) ;

				}

				admin.createTable(tableBuilder.build());

				System.out.println("Table create successful!");

			}

		} catch (Exception e) {

			e.printStackTrace();

		}

	}

	/**

	 * 删除表

	 * @param tabName

	 */

	public static void dropTable(String tabName) {

		try {

			TableName tableName = TableName.valueOf(tabName);

			admin.disableTable(tableName);

			admin.deleteTable(tableName);

			System.out.println("删除成功"+tabName);

			close();

		} catch (Exception e) {

			e.printStackTrace();

		}

	}

	/**

	 * 查询表列表

	 */

	public static void listTable() {

		try {

			TableName[] tbNames = admin.listTableNames();

			for (TableName tableName : tbNames) {

				System.out.println(tableName.getNameAsString());

			}

		} catch (Exception e) {

			e.printStackTrace();

		}

	}

	/**

	 * 新增一条记录

	 * @param tabName	表名

	 * @param rowKey	行键

	 * @param colFamily	列族

	 * @param column	列名

	 * @param value		单元值

	 */

	public static void insertRow(String tabName,String rowKey,String colFamily,String column,String value) {

		try {

			TableName tableName = TableName.valueOf(tabName);

			Table table = connection.getTable(tableName);

			Put put = new Put(rowKey.getBytes());

			put.addColumn(colFamily.getBytes(), column.getBytes(), value.getBytes());

			table.put(put);

			table.close();

			System.out.println("新增纪录成功！");

		} catch (Exception e) {

			e.printStackTrace();

		}

	}

	/**

	 * 删除行数据

	 * @param tabName		表名

	 * @param rowKey		行键

	 * @param colFamily		列族

	 * @param qualifier		列修饰符

	 */

	public static void deleteRow(String tabName,String rowKey,String colFamily,String qualifier) {

		try {

			TableName tableName = TableName.valueOf(tabName);

			Table table = connection.getTable(tableName);

			Delete delete = new Delete(rowKey.getBytes());

			// 删除指定列族的所有数据

			if(StringUtils.isNotEmpty(qualifier) && StringUtils.isNotEmpty(colFamily)) {

				delete.addColumn(colFamily.getBytes(),qualifier.getBytes());

			}else if(StringUtils.isNotEmpty(colFamily)) {

				delete.addFamily(colFamily.getBytes());

			}

			table.delete(delete);

			System.out.println("删除数据成功");

			table.close();

		} catch (Exception e) {

		}

	}

	/**

	 * 查询数据

	 * @param tabName

	 * @param rowKey

	 * @param colFamily

	 * @param qualifier

	 */

	public static void getData(String tabName,String rowKey,String colFamily,String qualifier) {

		try {

			TableName tableName = TableName.valueOf(tabName);

			Table table = connection.getTable(tableName);

			Get get = new Get(rowKey.getBytes());

			if(StringUtils.isNotEmpty(qualifier) && StringUtils.isNotEmpty(colFamily)) {

				get.addColumn(colFamily.getBytes(),qualifier.getBytes());

			}else if(StringUtils.isNotEmpty(colFamily)) {

				get.addFamily(colFamily.getBytes());

			}

			Result result = table.get(get);

			if(!result.isEmpty()) {

				showCell(result);

			}else {

				System.out.println("数据已经不存在");

			}

			table.close();

		} catch (Exception e) {

			e.printStackTrace();

		}

	}

	/**

	 * 格式化输出

	 * @param result

	 */

	private static void showCell(Result result){

		Cell[] cells = result.rawCells();

		for(Cell cell:cells){

			System.out.println("RowName:"+new String(CellUtil.cloneRow(cell))+" ");

			System.out.println("Timetamp:"+cell.getTimestamp()+" ");

			System.out.println("column Family:"+new String(CellUtil.cloneFamily(cell))+" ");

			System.out.println("row Name:"+new String(CellUtil.cloneQualifier(cell))+" ");

			System.out.println("value:"+new String(CellUtil.cloneValue(cell))+" ");

		}

	}

}