package com.hbase.HBaseAdmin;

import java.io.IOException;

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.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.RegionLocator;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;


/**
* @author:FengZhen
* @create:2018年9月6日
*/
public class CreateTable {

	private static String addr="HDP233,HDP232,HDP231";
	private static String port="2181";
	private static Connection connection;
	
	/**
	 * 获取连接
	 */
	public static void getConnection(){
		Configuration conf = HBaseConfiguration.create();

		conf.set("hbase.zookeeper.quorum",addr);
		conf.set("hbase.zookeeper.property.clientPort", port);
		try {
			connection = ConnectionFactory.createConnection(conf);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	/*
	 * 关闭连接
	 */
	public static void close() {
		/**
		 * close connection
		 **/
		if (connection != null) {
			try {
				connection.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
	
	public static void main(String[] args) {
		createTablePrePartition();
	}
	
	/**
	 * 建表
	 */
	public static void createTable() {
		getConnection();
		try {
			//获取admin实例
			Admin admin = connection.getAdmin();
			//创建表描述符
			HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf("test_create"));
			//添加列族描述符到表描述符中
			HColumnDescriptor columnDescriptor = new HColumnDescriptor(Bytes.toBytes("info"));
			tableDescriptor.addFamily(columnDescriptor);
			//调用create方法
			admin.createTable(tableDescriptor);
			//检查表是否可用
			boolean avail = admin.isTableAvailable(TableName.valueOf("test_create"));
			System.out.println("Table available: " + avail);
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			close();
		}
	}
	
	//---------------------------------------通过预分区建表-----------------------------------------
	/**
	 * 	Printing regions of table: test_pre_partition1
		[1]start key: ,end key:1
		[2]start key:1 ,end key:13
		[3]start key:13 ,end key:25
		[4]start key:25 ,end key:37
		[5]start key:37 ,end key:49
		[6]start key:49 ,end key:61
		[7]start key:61 ,end key:73
		[8]start key:73 ,end key:85
		[9]start key:85 ,end key:100
		[10]start key:100 ,end key:
		Printing regions of table: test_pre_partition2
		[1]start key: ,end key:A
		[2]start key:A ,end key:D
		[3]start key:D ,end key:G
		[4]start key:G ,end key:K
		[5]start key:K ,end key:O
		[6]start key:O ,end key:T
		[7]start key:T ,end key:
	 */
	/**
	 * 打印表中region信息
	 * @param tableName
	 * @throws IOException
	 */
	public static void printTableRegions(String tableName) throws IOException {
		System.out.println("Printing regions of table: " + tableName);
		//返回表中所有region的起始行键与终止行键列表
		RegionLocator regionLocator = connection.getRegionLocator(TableName.valueOf(tableName));
		//获取所有region的边界。
		//第一个region的起始行键与最后一个region的终止行键都是空字节，这是HBase中默认的规则
		//起始和终止行键都是已经计算好的，或是提供给用户的拆分键。
		//需要注意的是，前一个region的终止行键与后一个region的起始行键是串联起来的
		//终止行键不包含在前一个region中，而是作为起始行键包含在后一个region中。
		Pair<byte[][], byte[][]> pair = regionLocator.getStartEndKeys();
		for(int n = 0; n < pair.getFirst().length; n++) {
			byte[] sk = pair.getFirst()[n];
			byte[] ek = pair.getSecond()[n];
			System.out.println("[" + (n + 1) + "]" + 
			"start key:" + (sk.length == 8 ? Bytes.toLong(sk) : Bytes.toStringBinary(sk)) + 
			" ,end key:" + (ek.length == 8 ? Bytes.toLong(ek) : Bytes.toStringBinary(ek)));
		}
	}
	/**
	 * 通过预分区的方式建表
	 * @throws IOException 
	 */
	public static void createTablePrePartition() {
		getConnection();
		String tableName = "test_pre_partition1";
		String tableName2 = "test_pre_partition2";
		try {
			//获取admin实例
			Admin admin = connection.getAdmin();
			//创建表描述符
			HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
			//添加列族描述符到表描述符中
			HColumnDescriptor columnDescriptor = new HColumnDescriptor(Bytes.toBytes("info"));
			tableDescriptor.addFamily(columnDescriptor);
			//调用create方法,同时设置region边界。
			//能够以特定数量拆分特定起始行键和特定终止行键，并创建表。
			//startKey必须小于endKey，并且numRegions需要大于等于3，否则会抛出异常，这样才能确保region有最小的集合
			//此方法使用Bytes.split()方法计算region边界，然后将计算得到的边界作为已拆分边界列表，并调用createTable(final HTableDescriptor desc, byte[][] splitKeys)方法
			admin.createTable(tableDescriptor, Bytes.toBytes(1L), Bytes.toBytes(100L), 10);
			printTableRegions(tableName); 
			
			//创建表中region的拆分行键
			byte[][] regions = new byte[][] {
				Bytes.toBytes("A"),
				Bytes.toBytes("D"),
				Bytes.toBytes("G"),
				Bytes.toBytes("K"),
				Bytes.toBytes("O"),
				Bytes.toBytes("T")
			};
			tableDescriptor.setName(TableName.valueOf(tableName2));
			//使用新表明和region的已拆分键值列表作为参数调用建表命令
			//使用已拆分行键的集合：使用了已经拆分好的region边界列表，因此结果都是与预期相符的。
			admin.createTable(tableDescriptor, regions);
			printTableRegions(tableName2);
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			close();
		}
	}
}

还有createTableAsync方法，这个方法使用表描述符和预拆分的region边界作为参数，并进行异步建表，但执行过程与createTable殊途同归

同步模式仅仅是异步模式的简单封装，增加了不断检查这个任务是否已经完成的循环操作。