package hadoop;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
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.TableDescriptor;
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.util.Bytes;

public class CRUD {

	private static final String NM_BAOSHAOMING = "baoshaoming";
	private static final String COL_STUDENT_ID = "student_id";
	private static final String COL_CLASS = "class";
	private static final String COL_UNDERSTANDING = "understanding";
	private static final String COL_PROGRAMMING = "programming";
	private static final String CF_SCORE = "score";
	private static final String CF_INFO = "info";

	public static void main(String[] args) throws Exception {
		try {
			createTable();
		} catch (Exception e) {
			e.printStackTrace();
			System.out.println(e.getMessage());
		}
	}

	public static void createTable() throws Exception {
		Configuration config = HBaseConfiguration.create();
		config.set("hbase.zookeeper.quorum", "jikehadoop02");

		Connection connection = null;
		Admin admin = null;

		connection = ConnectionFactory.createConnection(config);
		admin = connection.getAdmin();
		boolean isNameSpaceCreated = false;
		NamespaceDescriptor[] namespaceDescriptors = admin.listNamespaceDescriptors();
		if (namespaceDescriptors != null && namespaceDescriptors.length > 0) {
			for (NamespaceDescriptor nm : namespaceDescriptors) {
				if (nm.getName().equals(NM_BAOSHAOMING)) {
					isNameSpaceCreated = true;
					break;
				}
			}
		}
		if (!isNameSpaceCreated) {
			// 创建namespace描述器建造者
			NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(NM_BAOSHAOMING).build();
			// 创建namespace
			admin.createNamespace(namespaceDescriptor);
		}
		String tableName = NM_BAOSHAOMING + ":student";
		if (!admin.isTableAvailable(TableName.valueOf(tableName))) {
			TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName));
			// 列族的构建器
			ColumnFamilyDescriptorBuilder cf1Builder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(CF_INFO));
			// 设置列族的属性
			cf1Builder.setTimeToLive(300);
			cf1Builder.setMaxVersions(3);
			ColumnFamilyDescriptor cf1 = cf1Builder.build();
			ColumnFamilyDescriptorBuilder cf2Builder = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(CF_SCORE));
			ColumnFamilyDescriptor cf2 = cf2Builder.build();
			List<ColumnFamilyDescriptor> ls = new ArrayList<>();
			ls.add(cf1);
			ls.add(cf2);
			// 添加列族
			tableDescriptorBuilder.setColumnFamilies(ls);
			TableDescriptor descriptor = tableDescriptorBuilder.build();
			admin.createTable(descriptor);
			admin.close();
		}

		List<Put> putList = new ArrayList<>();
		if (!admin.tableExists(TableName.valueOf(tableName))) {
			System.err.println("the " + tableName + " is not exist");
			System.exit(1);
		}

		deleteData(TableName.valueOf(tableName),"Tom",connection);
		deleteData(TableName.valueOf(tableName),"Jerry",connection);
		deleteData(TableName.valueOf(tableName),"Jack",connection);
		deleteData(TableName.valueOf(tableName),"鲍少明",connection);
		System.out.println("delete data done");
		Put put1 = new Put(Bytes.toBytes("Tom"));
		put1.addColumn(Bytes.toBytes(CF_INFO), Bytes.toBytes(COL_STUDENT_ID), Bytes.toBytes("20210000000001"));
		put1.addColumn(Bytes.toBytes(CF_INFO), Bytes.toBytes(COL_CLASS), Bytes.toBytes("1"));
		put1.addColumn(Bytes.toBytes(CF_SCORE), Bytes.toBytes(COL_UNDERSTANDING), Bytes.toBytes("75"));
		put1.addColumn(Bytes.toBytes(CF_SCORE), Bytes.toBytes(COL_PROGRAMMING), Bytes.toBytes("82"));
		putList.add(put1);
		Put put2 = new Put(Bytes.toBytes("Jerry"));
		put2.addColumn(Bytes.toBytes(CF_INFO), Bytes.toBytes(COL_STUDENT_ID), Bytes.toBytes("20210000000002"));
		put2.addColumn(Bytes.toBytes(CF_INFO), Bytes.toBytes(COL_CLASS), Bytes.toBytes("1"));
		put2.addColumn(Bytes.toBytes(CF_SCORE), Bytes.toBytes(COL_UNDERSTANDING), Bytes.toBytes("85"));
		put2.addColumn(Bytes.toBytes(CF_SCORE), Bytes.toBytes(COL_PROGRAMMING), Bytes.toBytes("67"));
		putList.add(put2);
		Put put3 = new Put(Bytes.toBytes("Jack"));
		put3.addColumn(Bytes.toBytes(CF_INFO), Bytes.toBytes(COL_STUDENT_ID), Bytes.toBytes("20210000000003"));
		put3.addColumn(Bytes.toBytes(CF_INFO), Bytes.toBytes(COL_CLASS), Bytes.toBytes("2"));
		put3.addColumn(Bytes.toBytes(CF_SCORE), Bytes.toBytes(COL_UNDERSTANDING), Bytes.toBytes("80"));
		put3.addColumn(Bytes.toBytes(CF_SCORE), Bytes.toBytes(COL_PROGRAMMING), Bytes.toBytes("80"));
		putList.add(put3);
		Put put5 = new Put(Bytes.toBytes("Rose"));
		put5.addColumn(Bytes.toBytes(CF_INFO), Bytes.toBytes(COL_STUDENT_ID), Bytes.toBytes("20210000000004"));
		put5.addColumn(Bytes.toBytes(CF_INFO), Bytes.toBytes(COL_CLASS), Bytes.toBytes("2"));
		put5.addColumn(Bytes.toBytes(CF_SCORE), Bytes.toBytes(COL_UNDERSTANDING), Bytes.toBytes("60"));
		put5.addColumn(Bytes.toBytes(CF_SCORE), Bytes.toBytes(COL_PROGRAMMING), Bytes.toBytes("61"));
		putList.add(put5);
		Put put4 = new Put(Bytes.toBytes("鲍少明"));
		put4.addColumn(Bytes.toBytes(CF_INFO), Bytes.toBytes(COL_STUDENT_ID), Bytes.toBytes("G20210698020030"));
		put4.addColumn(Bytes.toBytes(CF_INFO), Bytes.toBytes(COL_CLASS), Bytes.toBytes("5"));
		put4.addColumn(Bytes.toBytes(CF_SCORE), Bytes.toBytes(COL_UNDERSTANDING), Bytes.toBytes("100"));
		put4.addColumn(Bytes.toBytes(CF_SCORE), Bytes.toBytes(COL_PROGRAMMING), Bytes.toBytes("100"));
		putList.add(put4);
		Table table = connection.getTable(TableName.valueOf(tableName));
		table.put(putList);
		table.close();
		System.out.println("put data done");
		System.out.println("query 鲍少明 start");
		Get get = new Get(Bytes.toBytes("鲍少明"));
        Result set = table.get(get);
        Cell[] cells = set.rawCells();
        for (Cell cell : cells) {
            System.out.println(Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()) + "::" +
                    Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
        } 
        System.out.println(Bytes.toString(set.getValue(Bytes.toBytes(CF_INFO),Bytes.toBytes(COL_STUDENT_ID)))); 
        table.close();
        System.out.println("query data done");

	}
	
	public static void deleteData(TableName tableName, String rowKey, Connection conn) throws IOException {
        Table table = conn.getTable(tableName);
 
        Delete delete = new Delete(Bytes.toBytes(rowKey));
 
        table.delete(delete);
        table.close();
    }
 

	
	

}