Hbase(六) hbase Java API

1、

幾個主要 Hbase API 類和數據模型之間的對應關係：

一、 HBaseAdmin
關係： org.apache.hadoop.hbase.client.HBaseAdmin
做用：提供了一個接口來管理 HBase 數據庫的表信息。它提供的方法包括：建立表，刪 除表，列出表項，使表有效或無效，以及添加或刪除表列族成員等。

二、 HBaseConfiguration
關係： org.apache.hadoop.hbase.HBaseConfiguration
做用：對 HBase 進行配置

三、 HTableDescriptor
關係： org.apache.hadoop.hbase.HTableDescriptor
做用：包含了表的名字極其對應表的列族

四、 HColumnDescriptor
關係： org.apache.hadoop.hbase.HColumnDescriptor
做用：維護着關於列族的信息，例如版本號，壓縮設置等。它一般在建立表或者爲表添 加列族的時候使用。列族被建立後不能直接修改，只能經過刪除而後從新建立的方式。
列族被刪除的時候，列族裏面的數據也會同時被刪除。

五、 HTable
關係： org.apache.hadoop.hbase.client.HTable
做用：能夠用來和 HBase 表直接通訊。此方法對於更新操做來講是非線程安全的。

六、 Put
關係： org.apache.hadoop.hbase.client.Put
做用：用來對單個行執行添加操做

七、 Get
關係： org.apache.hadoop.hbase.client.Get
做用：用來獲取單個行的相關信息

八、 Result
關係： org.apache.hadoop.hbase.client.Result
做用：存儲 Get 或者 Scan 操做後獲取表的單行值。使用此類提供的方法能夠直接獲取值 或者各類 Map 結構（ key-value 對）

 2、具體增刪改查    代碼具體實現：

package HbaseDome;



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.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
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.protobuf.generated.ZooKeeperProtos.Table;
import org.apache.hadoop.hbase.util.Bytes;



public class Hbasedome implements HBaseDemoInterface{

	
	static Configuration conf =null;
	private static final String ZKconnect="192.168.123.212:2181,192.168.123.213:2181,192.168.123.214:2181";
	static{
		conf=HBaseConfiguration.create();
		conf.set("hbase.zookeeper.quorum", ZKconnect);
	}
//	static String tableName="student";
//	static String[] family={"lie01","lie02"};
	
	
	public static void main(String[] args) {
		Hbasedome a =new Hbasedome();
		 String tableName="student11";
		 String[] family={"lie01","lie02"};
		try {
			HTableDescriptor htds =new HTableDescriptor(tableName);
			for(int z=0;z<family.length;z++){
				HColumnDescriptor h=new HColumnDescriptor(family[z]);
				htds.addFamily(h);
			}
//			a.descTable("table03");
//			a.createTable(tableName, htds);
//			a.descTable("table03");
//			a.getAllTables();
//			a.createTable(tableName,family);
//			a.getResult("table03", "usr001");
//			a.dropTable("user1");
//			a.getAllTables();
//			a.putData("table03", "usr005", "liezu01", "name", "liu");
//			a.getResult("table03", "usr001");
//			a.getResultScann("table03");
//			a.getResultScann("table03","");
			
			Result result = a.getResult("table03", "usr001");
			System.out.println(result.toString());
			List<Cell> cells = result.listCells();
			for (int i = 0; i < cells.size(); i++) {
				Cell cell = cells.get(i);
				System.out.println(cell.toString());
	//			printCell(cell);
			}
		
//			List<KeyValue> list = result.list();
//			for (int i = 0; i < list.size(); i++) {
//				KeyValue kv = list.get(i);
//				printKeyValye(kv);
//			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		
	}
	public static void printKeyValye(KeyValue kv) {
		System.out.println(Bytes.toString(kv.getRow()) + "\t" + Bytes.toString(kv.getFamily()) + "\t" + Bytes.toString(kv.getQualifier()) + "\t" + Bytes.toString(kv.getValue()) + "\t" + kv.getTimestamp());
	}
	public static void printCell(Cell cell) {
		System.out.println(Bytes.toString(cell.getRow()) + "\t" + Bytes.toString(cell.getFamily()) + "\t" + Bytes.toString(cell.getQualifier()) + "\t" + Bytes.toString(cell.getValue()) + "\t" + cell.getTimestamp());
	}
	//建立表
	@Override
	public void createTable(String tableName, String[] family) throws Exception {
		HBaseAdmin admin=new HBaseAdmin(conf);
		HTableDescriptor desc =new HTableDescriptor(tableName);
		
		for(int i=0;i<family.length;i++){
			desc.addFamily(new HColumnDescriptor(family[i]));
			System.out.println("11111111111"+family[i]);
		}
		if(admin.tableExists(tableName)){
			System.out.println("表已經存在，別瞎輸行嗎");
//			System.exit(0);
		}else{
			admin.createTable(desc);
			System.out.println("表建立成功");
		}
	}

	//建立表
	@Override
	public void createTable(String tableName, HTableDescriptor htds) throws Exception {
		HBaseAdmin admin=new HBaseAdmin(conf);
		boolean tableExists1 = admin.tableExists(Bytes.toBytes(tableName));
		System.out.println(tableExists1 ? "表已存在" : "表不存在");
		admin.createTable(htds);
		boolean tableExists = admin.tableExists(Bytes.toBytes(tableName));
		System.out.println(tableExists ? "建立表成功" : "建立失敗");
	}

	
	@Override
	public void descTable(String tableName) throws Exception {
		HBaseAdmin admin=new HBaseAdmin(conf);
		HTable table=new HTable(conf, tableName);
		HTableDescriptor desc =table.getTableDescriptor();
		HColumnDescriptor[] columnFamilies = desc.getColumnFamilies();
	
		for(HColumnDescriptor t:columnFamilies){
			System.out.println(Bytes.toString(t.getName()));
		}
		
	}

	//// 這種方式是替換該表tableName的全部列簇
	@Override
	public void modifyTable(String tableName) throws Exception {
		HBaseAdmin admin=new HBaseAdmin(conf);
		HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(tableName));
		htd.addFamily(new HColumnDescriptor(Bytes.toBytes("cf3")));
		htd.addFamily(new HColumnDescriptor(Bytes.toBytes("cf2")));
		admin.modifyTable(tableName, htd);

		// 刪除該表tableName當中的特定的列簇
		// admin.deleteColumn(tableName, "cf3");

		System.out.println("修改爲功");
		
	}

	@Override
	public void getAllTables() throws Exception {
		HBaseAdmin admin =new HBaseAdmin(conf);
		
		String[] tableNames = admin.getTableNames();
		for(int i=0;i<tableNames.length;i++){
			System.out.println(tableNames[i]);
		}
	}

	//更新數據  插入數據
	@Override
	public void putData(String tableName, String rowKey, String familyName, String columnName, String value)
			throws Exception {
		HTable htable=new HTable(conf, Bytes.toBytes(tableName));
		Put put=new Put(Bytes.toBytes(rowKey));
		put.add(Bytes.toBytes(familyName), Bytes.toBytes(columnName), Bytes.toBytes(value));
		htable.put(put);
		
	}

	//爲表添加數據
	@Override
	public void addData(String tableName, String rowKey, String[] column1, String[] value1, String[] column2,
			String[] value2) throws Exception {
		
		Put put=new Put(Bytes.toBytes(rowKey));
		HTable htable=new HTable(conf, Bytes.toBytes(tableName));
		HColumnDescriptor[] columnFamilies = htable.getTableDescriptor().getColumnFamilies();
		for(int i=0;i<=columnFamilies.length;i++){
			String nameAsString = columnFamilies[i].getNameAsString();
			if(nameAsString.equals("lie01")){
				for(int j=0;j<column1.length;j++){
					put.add(Bytes.toBytes(nameAsString), Bytes.toBytes(column1[j]),Bytes.toBytes(value1[j]));
				}
			}
			if(nameAsString.equals("lie02")){
				for(int j=0;j<column2.length;j++){
					put.add(Bytes.toBytes(nameAsString), Bytes.toBytes(column2[j]),Bytes.toBytes(value2[j]));
				}
			}
			
		}
		htable.put(put);
		System.out.println("addData ok!");
	}

	//根據rowkey 查詢
	@Override
	public Result getResult(String tableName, String rowKey) throws Exception {
		Get get=new Get(Bytes.toBytes(rowKey));
		HTable htable=new HTable(conf, Bytes.toBytes(tableName));
		Result result=htable.get(get);
//		for(KeyValue k:result.list()){
//			System.out.println(Bytes.toString(k.getFamily()));
//			System.out.println(Bytes.toString(k.getQualifier()));
//			System.out.println(Bytes.toString(k.getValue()));
//			System.out.println(k.getTimestamp());
//		}
		return result;
	}

	//查詢指定的某列
	@Override
	public Result getResult(String tableName, String rowKey, String familyName, String columnName) throws Exception {
		Get get=new Get(Bytes.toBytes(rowKey));
		HTable htable=new HTable(conf, Bytes.toBytes(tableName));
		get.addColumn(Bytes.toBytes(familyName),Bytes.toBytes(columnName));
		Result result=htable.get(get);
		for(KeyValue k:result.list()){
			System.out.println(Bytes.toString(k.getFamily()));
			System.out.println(Bytes.toString(k.getQualifier()));
			System.out.println(Bytes.toString(k.getValue()));
			System.out.println(k.getTimestamp());
		}
		return result;
	}

	
	//遍歷查詢表
	@Override
	public ResultScanner getResultScann(String tableName) throws Exception {
	
		Scan scan=new Scan();
		ResultScanner rs =null;
		HTable htable=new HTable(conf, tableName);
		try{
			rs=htable.getScanner(scan);
			for(Result r: rs){
				for(KeyValue kv:r.list()){
	
					System.out.println(Bytes.toString(kv.getRow()));
					System.out.println(Bytes.toString(kv.getFamily()));
					System.out.println(Bytes.toString(kv.getQualifier()));
					System.out.println(Bytes.toString(kv.getValue()));
					System.out.println(kv.getTimestamp());
				}
			}
		}finally{
			rs.close();
		}
		return rs;
	}

	@Override
	public ResultScanner getResultScann(String tableName, Scan scan) throws Exception {
		
		ResultScanner rs =null;
		HTable htable=new HTable(conf, tableName);
		try{
			rs=htable.getScanner(scan);
			for(Result r: rs){
				for(KeyValue kv:r.list()){
	
					System.out.println(Bytes.toString(kv.getRow()));
					System.out.println(Bytes.toString(kv.getFamily()));
					System.out.println(Bytes.toString(kv.getQualifier()));
					System.out.println(Bytes.toString(kv.getValue()));
					System.out.println(kv.getTimestamp());
				}
			}
		}finally{
			rs.close();
		}		
		return rs;
	}

	//查詢表中的某一列
	@Override
	public Result getResultByColumn(String tableName, String rowKey, String familyName, String columnName)
			throws Exception {
		
		
		HTable htable=new HTable(conf, tableName);
		Get get=new Get(Bytes.toBytes(rowKey));
		get.addColumn(Bytes.toBytes(familyName),Bytes.toBytes(columnName));
		Result result=htable.get(get);
		for(KeyValue kv: result.list()){

			System.out.println(Bytes.toString(kv.getFamily()));
			System.out.println(Bytes.toString(kv.getQualifier()));
			System.out.println(Bytes.toString(kv.getValue()));
			System.out.println(kv.getTimestamp());
			
		}
		return result;
	}

	
	//查詢某列數據的某個版本
	@Override
	public Result getResultByVersion(String tableName, String rowKey, String familyName, String columnName,
			int versions) throws Exception {
	
		HTable htable=new HTable(conf, tableName);
		Get get =new Get(Bytes.toBytes(rowKey));
		get.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName));
	    get.setMaxVersions(versions);
	    Result result=htable.get(get);
	    
	    for(KeyValue kv: result.list()){

			System.out.println(Bytes.toString(kv.getFamily()));
			System.out.println(Bytes.toString(kv.getQualifier()));
			System.out.println(Bytes.toString(kv.getValue()));
			System.out.println(kv.getTimestamp());
			
		}

		return result;
	}

    //刪除指定某列
	@Override
	public void deleteColumn(String tableName, String rowKey, String falilyName, String columnName) throws Exception {

		HTable htable=new HTable(conf, tableName);
//		Delete delete1=new Delete(Bytes.toBytes(rowKey));
		Delete de =new Delete(Bytes.toBytes(rowKey));
		de.deleteColumn(Bytes.toBytes(falilyName), Bytes.toBytes(columnName));
	    htable.delete(de);
	}

	
	//刪除指定的某個rowkey
	@Override
	public void deleteColumn(String tableName, String rowKey) throws Exception {
		HTable htable=new HTable(conf, tableName);

		Delete de =new Delete(Bytes.toBytes(rowKey));
		 htable.delete(de);
		
	}

	//讓該表失效
	@Override
	public void disableTable(String tableName) throws Exception {
		HBaseAdmin admin=new HBaseAdmin(conf);
		admin.disableTable(tableName);
		
	}

	//刪除表
	@Override
	public void dropTable(String tableName) throws Exception {
		
		HBaseAdmin admin=new HBaseAdmin(conf);
		admin.disableTable(tableName);
		admin.deleteTable(tableName);
		
	}

}

　　

　　

 

package com.ghgj.hbase.test1610;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
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.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
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 HBaseAPIDemo1610 implements HBaseDemoInterface {

	private static final String ROWKEY = "p001";
	private static final String ROWKEY2 = "p002";
	private static final String FAMILY1 = "cf1";
	private static final String FAMILY2 = "cf2";
	private static final String KEY = "name";
	private static final String VALUE = "huangbo";

	private static final String TABLE_NAME = "person";
	private static final String[] COLUMN_FAMILY = new String[] { FAMILY1, FAMILY2 };

	static Configuration conf = null;
	static HBaseAdmin admin = null;
	static HTable table = null;

	static {
		try {
			conf = HBaseConfiguration.create();
			conf.set("hbase.zookeeper.quorum", "hadoop03:2181,hadoop04:2181,hadoop05:2181");
			admin = new HBaseAdmin(conf);
			table = new HTable(conf, TABLE_NAME);

		} catch (IOException e) {
			// e.printStackTrace();
			System.out.println("報錯");
		}
	}

	public static void main(String[] args) throws Exception {
		HBaseAPIDemo1610 hbase = new HBaseAPIDemo1610();

		// 測試建立表
		hbase.createTable(TABLE_NAME, COLUMN_FAMILY);

		// 測試建立表
		HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(TABLE_NAME));
		for (int i = 0; i < COLUMN_FAMILY.length; i++) {
			HColumnDescriptor cf1 = new HColumnDescriptor(COLUMN_FAMILY[i]);
			htd.addFamily(cf1);
		}
		hbase.createTable(TABLE_NAME, htd);

		// 查看錶屬性
		hbase.descTable(TABLE_NAME);

		// 查詢全部的表
		hbase.getAllTables();

		// 測試修改表
		hbase.modifyTable(TABLE_NAME);

		// 插入數據
		hbase.putData(TABLE_NAME, ROWKEY, FAMILY1, KEY, VALUE);

		// 測試插入一堆數據
		String[] column1 = new String[] { "name1", "age", "province" };
		String[] value1 = new String[] { "huangbo", "33", "xinjiang" };
		String[] column2 = new String[] { "gender" };
		String[] value2 = new String[] { "male" };
		hbase.addData(TABLE_NAME, ROWKEY2, column1, value1, column2, value2);

		// 經過rowkey查詢數據
		Result result = hbase.getResult(TABLE_NAME, ROWKEY2);
		System.out.println(result.toString());
		List<KeyValue> list = result.list();
		for (int i = 0; i < list.size(); i++) {
			KeyValue kv = list.get(i);
			printKeyValye(kv);
		}

		// 經過rowkey, family, province查詢數據
		Result result1 = hbase.getResult(TABLE_NAME, ROWKEY2, FAMILY1, "province");
		List<Cell> cells = result1.listCells();
		for (int i = 0; i < cells.size(); i++) {
			Cell cell = cells.get(i);
			printCell(cell);
		}

		// 掃描全表數據
		ResultScanner resultScann = hbase.getResultScann(TABLE_NAME);
		printResultScanner(resultScann);
		
		/*Iterator<Result> iterator = resultScann.iterator();
		while(iterator.hasNext()){
			Result next = iterator.next();
		}*/

		// 經過scan掃描全表數據，scan中能夠加入一些過濾條件
		Scan scan = new Scan();
		scan.setStartRow(Bytes.toBytes("user"));
		scan.setStopRow(Bytes.toBytes("zk002"));
		scan.setTimeRange(1488252774189l, 1488252774191l);
		ResultScanner resultScann1 = hbase.getResultScann(TABLE_NAME, scan);
		printResultScanner(resultScann1);

		// 兩種方式查詢最大版本數的hbase數據
		Result resultByVersion = hbase.getResultByVersion(TABLE_NAME, ROWKEY, FAMILY1, "name", 3);
		printResult(resultByVersion);
		System.out.println("-------------------");
		ResultScanner rs = hbase.getResultByVersion(ROWKEY, FAMILY1, "name", 3);
		printResultScanner(rs);

		// 刪除表
		hbase.dropTable(TABLE_NAME);
	}

	public static void printResultScanner(ResultScanner resultScann) {
		for (Result result : resultScann) {
			printResult(result);
		}
	}

	public static void printResult(Result result) {
		List<Cell> cells = result.listCells();
		for (int i = 0; i < cells.size(); i++) {
			Cell cell = cells.get(i);
			printCell(cell);
		}
	}

	public static void printCell(Cell cell) {
		System.out.println(Bytes.toString(cell.getRow()) + "\t" + Bytes.toString(cell.getFamily()) + "\t" + Bytes.toString(cell.getQualifier()) + "\t" + Bytes.toString(cell.getValue()) + "\t" + cell.getTimestamp());
	}

	public static void printKeyValye(KeyValue kv) {
		System.out.println(Bytes.toString(kv.getRow()) + "\t" + Bytes.toString(kv.getFamily()) + "\t" + Bytes.toString(kv.getQualifier()) + "\t" + Bytes.toString(kv.getValue()) + "\t" + kv.getTimestamp());
	}

	// create 'tablename','cf1','cf2'
	@Override
	public void createTable(String tableName, String[] family) throws Exception {
		HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(tableName));
		for (int i = 0; i < family.length; i++) {
			HColumnDescriptor cf1 = new HColumnDescriptor(family[i]);
			htd.addFamily(cf1);
		}
		admin.createTable(htd);
		boolean tableExists = admin.tableExists(Bytes.toBytes(tableName));
		System.out.println(tableExists ? "建立表成功" : "建立失敗");
	}

	@Override
	public void createTable(String tableName, HTableDescriptor htd) throws Exception {
		admin.createTable(htd);
		boolean tableExists = admin.tableExists(Bytes.toBytes(tableName));
		System.out.println(tableExists ? "建立表成功" : "建立失敗");
	}

	// desc 'person'
	@Override
	public void descTable(String tableName) throws Exception {
		HTableDescriptor tableDescriptor = table.getTableDescriptor();
		HColumnDescriptor[] columnFamilies = tableDescriptor.getColumnFamilies();
		for (HColumnDescriptor hcd : columnFamilies) {
			// System.out.println(hcd.toString()+"\t");
			System.out.println(Bytes.toString(hcd.getName()));
		}
	}

	@Override
	public void modifyTable(String tableName) throws Exception {
		// 這種方式是替換該表tableName的全部列簇
		HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(tableName));
		htd.addFamily(new HColumnDescriptor(Bytes.toBytes("cf3")));
		htd.addFamily(new HColumnDescriptor(Bytes.toBytes("cf2")));
		admin.modifyTable(tableName, htd);

		// 刪除該表tableName當中的特定的列簇
		// admin.deleteColumn(tableName, "cf3");

		System.out.println("修改爲功");
	}

	// list
	@Override
	public void getAllTables() throws Exception {
		TableName[] listTableNames = admin.listTableNames();
		for (TableName tn : listTableNames) {
			System.out.println(tn.toString());
		}
	}

	// put 'tablename','rowkey','familyname:key','value'
	@Override
	public void putData(String tableName, String rowKey, String familyName, String columnName, String value) throws Exception {
		// HTable table = new HTable(conf, tableName);
		Put put = new Put(Bytes.toBytes(rowKey));
		put.add(Bytes.toBytes(familyName), Bytes.toBytes(columnName), Bytes.toBytes(value));
		table.put(put);
		System.out.println("插入成功");
	}

	/**
	 * @param tableName
	 *            表名
	 * @param rowKey
	 *            rowkey
	 * @param column1
	 *            第一個列簇的key數組
	 * @param value1
	 *            第一個列簇的value數組，key數組和value數組長度必須同樣
	 * @param column2
	 *            第二列簇的key數組
	 * @param value2
	 *            第二個列簇的values數組， 同上同理
	 * @throws Exception
	 */
	@Override
	public void addData(String tableName, String rowKey, String[] column1, String[] value1, String[] column2, String[] value2) throws Exception {
		List<Put> puts = new ArrayList<Put>();

		for (int i = 0; i < column1.length; i++) {
			Put put = new Put(Bytes.toBytes(rowKey));
			put.add(Bytes.toBytes(FAMILY1), Bytes.toBytes(column1[i]), Bytes.toBytes(value1[i]));
			puts.add(put);
		}

		for (int i = 0; i < column2.length; i++) {
			Put put = new Put(Bytes.toBytes(rowKey));
			put.add(Bytes.toBytes(FAMILY2), Bytes.toBytes(column2[i]), Bytes.toBytes(value2[i]));
			puts.add(put);
		}

		table.put(puts);
		System.out.println("插入一堆數據成功");
	}

	// get 'tablename','rowkey'
	@Override
	public Result getResult(String tableName, String rowKey) throws Exception {
		Get get = new Get(Bytes.toBytes(rowKey));
		Result result = table.get(get);
		return result;
	}

	@Override
	public Result getResult(String tableName, String rowKey, String familyName, String columnName) throws Exception {
		Get get = new Get(Bytes.toBytes(rowKey));
		get.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName));
		Result result = table.get(get);
		return result;
	}

	@Override
	public ResultScanner getResultScann(String tableName) throws Exception {
		Scan scan = new Scan();
		ResultScanner scanner = table.getScanner(scan);
		// ResultScanner scanner = table.getScanner(Bytes.toBytes(FAMILY2));
		// ResultScanner scanner = table.getScanner(Bytes.toBytes(FAMILY1),
		// Bytes.toBytes("name1"));
		return scanner;
	}

	@Override
	public ResultScanner getResultScann(String tableName, Scan scan) throws Exception {
		return table.getScanner(scan);
	}

	@Override
	public Result getResultByColumn(String tableName, String rowKey, String familyName, String columnName) throws Exception {
		return null;
	}

	// get 'person','p001',{COLUMNS => 'cf1:name', VERSIONS => 3}
	@Override
	public Result getResultByVersion(String tableName, String rowKey, String familyName, String columnName, int versions) throws Exception {
		Get get = new Get(Bytes.toBytes(rowKey));
		get.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName));
		get.setMaxVersions(versions);
		Result result = table.get(get);
		return result;
	}

	public ResultScanner getResultByVersion(String rowKey, String familyName, String columnName, int versions) throws Exception {
		Scan scan = new Scan(Bytes.toBytes(rowKey), Bytes.toBytes(rowKey));
		scan.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(columnName));
		scan.setMaxVersions(versions);
		ResultScanner scanner = table.getScanner(scan);
		return scanner;
	}

	@Override
	public void deleteColumn(String tableName, String rowKey, String falilyName, String columnName) throws Exception {

	}

	@Override
	public void deleteColumn(String tableName, String rowKey) throws Exception {

	}

	@Override
	public void disableTable(String tableName) throws Exception {
		admin.disableTable(tableName);
	}

	@Override
	public void dropTable(String tableName) throws Exception {
		try {
			admin.deleteTable(tableName);
		} catch (Exception e) {
			// e.printStackTrace();
			disableTable(tableName);
			admin.deleteTable(tableName);
			System.out.println("ssssssss");
		} finally {
			boolean tableExists = admin.tableExists(Bytes.toBytes(tableName));
			System.out.println(tableExists ? "刪除失敗" : "刪除成功");
		}
	}
}