hbase的基本操作

本文列举一些hbase的基本操作代码。

package allen.studyhbase;

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

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
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;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;

public class BaseTest {
	protected static Log log = LogFactory.getLog(BaseTest.class);

	protected static HBaseAdmin hbaseAdmin;
	protected static HTablePool pool;
	protected static String TableName = "allen_test";
	protected static byte[] TableNameBytes = Bytes.toBytes(TableName);
	protected static byte[] ColumnFamilyName = Bytes.toBytes("cf");

	protected static String QNameStr1 = "q1";
	protected static String QNameStr2 = "q2";
	protected static String QNameStr3 = "q3";

	protected static byte[] QName1 = Bytes.toBytes(QNameStr1);
	protected static byte[] QName2 = Bytes.toBytes(QNameStr2);
	protected static byte[] QName3 = Bytes.toBytes(QNameStr3);

	protected HTableInterface table;

	@BeforeClass
	public static void beforeClass() throws Exception {
		hbaseAdmin = CommonConfig.getHBaseAdmin();
		deleteTable();
		createTable();
	}

	@AfterClass
	public static void afterClass() throws Exception {
		deleteTable();
	}

	@Before
	public void before() throws Throwable {
		pool = CommonConfig.getHTablePool();
		table = pool.getTable(TableName);
		fillData();
	}

	@After
	public void after() throws Exception {
		try {
			// full scan.
			Scan scan = new Scan();
			ResultScanner resultScanner = table.getScanner(scan);

			List<byte[]> rows = new LinkedList<byte[]>();
			for (Result result = resultScanner.next(); result != null; result = resultScanner
					.next()) {
				rows.add(result.getRow());
			}

			for (byte[] row : rows) {
				table.delete(new Delete(row));
				log.info("delete " + Bytes.toString(row));
			}

			// return table to pool.
			table.close();

		} catch (IOException e) {
			throw new RuntimeException(e);
		}
	}

	private static void createTable() throws Exception {
		// create new table.
		HTableDescriptor tableDescriptor = new HTableDescriptor(TableName);
		tableDescriptor.addFamily(new HColumnDescriptor(ColumnFamilyName));
		hbaseAdmin.createTable(tableDescriptor);
	}

	private static void deleteTable() throws Exception {
		// delete table if table exist.
		if (hbaseAdmin.tableExists(TableName)) {
			// disable table before delete it.
			hbaseAdmin.disableTable(TableName);
			hbaseAdmin.deleteTable(TableName);
		}
	}

	String rowKeyStr1 = "allen_test_row1";
	String rowKeyStr2 = "allen_test_row2";
	String rowKeyStr3 = "allen_test_row3";
	String rowKeyStr4 = "allen_test_row4";

	byte[] rowKey1 = Bytes.toBytes(rowKeyStr1);
	byte[] rowKey2 = Bytes.toBytes(rowKeyStr2);
	byte[] rowKey3 = Bytes.toBytes(rowKeyStr3);
	byte[] rowKey4 = Bytes.toBytes(rowKeyStr4);

	private void fillData() throws Throwable {

		Put put = new Put(rowKey1);
		put.add(ColumnFamilyName, QName1, Bytes.toBytes(100L));
		put.add(ColumnFamilyName, QName2, Bytes.toBytes(100L));
		table.put(put);

		put = new Put(rowKey2);
		put.add(ColumnFamilyName, QName1, Bytes.toBytes(20L));
		put.add(ColumnFamilyName, QName2, Bytes.toBytes(200L));
		table.put(put);

		// set null case.
		put = new Put(rowKey3);
		put.add(ColumnFamilyName, QName1, null);
		put.add(ColumnFamilyName, QName2, null);
		table.put(put);

		// empty case.
		put = new Put(rowKey4);
		put.add(ColumnFamilyName, QName3, Bytes.toBytes("test"));
		table.put(put);
	}

}

package allen.studyhbase;



import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.hbase.HBaseConfiguration;

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

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



/**

 * CommonConfig.

 * */

public class CommonConfig {



	private static HTablePool pool;

	private static HBaseAdmin hbaseAdmin;

	private static Configuration conf;



	static {

		try {

			conf = HBaseConfiguration.create();

			pool = new HTablePool(conf, 50);

			hbaseAdmin = new HBaseAdmin(conf);

		} catch (Exception e) {

			throw new RuntimeException(e);

		}

	};



	public static Configuration getConfiguration() {

		return conf;

	}



	public static HBaseAdmin getHBaseAdmin() {

		return hbaseAdmin;

	}



	public static HTablePool getHTablePool() {

		return pool;

	}

}

package allen.studyhbase;

import java.util.HashSet;
import java.util.Set;

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.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;

import org.apache.hadoop.hbase.util.Bytes;

import org.junit.Assert;

import org.junit.Test;

public class TestHbaseOp extends BaseTest {

	@Test
	public void testDelete() throws Exception {

		Get get = new Get(rowKey1);
		Result result = table.get(get);
		Assert.assertTrue(!result.isEmpty());

		Delete delete = new Delete(rowKey1);
		table.delete(delete);

		get = new Get(rowKey1);
		result = table.get(get);
		Assert.assertTrue(result.isEmpty());

	}

	@Test
	public void testDeleteNotExistRow() throws Exception {
		byte[] rowKey = Bytes.toBytes("allen_test_row");
		Delete delete = new Delete(rowKey);
		table.delete(delete);
	}

	@Test
	public void testScan_01() throws Exception {

		Set<String> resultRowKeys = new HashSet<String>();
		Scan scan = new Scan(rowKey1, rowKey2);
		ResultScanner resultScanner = table.getScanner(scan);
		for (Result result = resultScanner.next(); result != null; result = resultScanner
				.next()) {
			resultRowKeys.add(Bytes.toString(result.getRow()));
		}

		Assert.assertTrue(resultRowKeys.size() == 1);
		Assert.assertTrue(resultRowKeys.contains(rowKeyStr1));
	}

	@Test
	public void testScan_02() throws Exception {
		Set<String> resultRowKeys = new HashSet<String>();

		Scan scan = new Scan(rowKey1);
		ResultScanner resultScanner = table.getScanner(scan);
		for (Result result = resultScanner.next(); result != null; result = resultScanner
				.next()) {
			resultRowKeys.add(Bytes.toString(result.getRow()));
		}

		Assert.assertTrue(resultRowKeys.size() == 4);
		Assert.assertTrue(resultRowKeys.contains(rowKeyStr1));
		Assert.assertTrue(resultRowKeys.contains(rowKeyStr2));
		Assert.assertTrue(resultRowKeys.contains(rowKeyStr3));
		Assert.assertTrue(resultRowKeys.contains(rowKeyStr4));
	}

	@Test
	public void testScan_03() throws Exception {
		Set<String> resultRowKeys = new HashSet<String>();

		Scan scan = new Scan(rowKey1);
		scan.addColumn(ColumnFamilyName, QName1);
		ResultScanner resultScanner = table.getScanner(scan);
		for (Result result = resultScanner.next(); result != null; result = resultScanner
				.next()) {
			resultRowKeys.add(Bytes.toString(result.getRow()));
		}

		Assert.assertTrue(resultRowKeys.size() == 3);
		Assert.assertTrue(resultRowKeys.contains(rowKeyStr1));
		Assert.assertTrue(resultRowKeys.contains(rowKeyStr2));
		Assert.assertTrue(resultRowKeys.contains(rowKeyStr3));

	}

	@Test
	public void testCheckAndPut() throws Exception {
		byte[] rowKey = Bytes.toBytes("allen_test_row");
		Put put = new Put(rowKey);
		put.add(ColumnFamilyName, QName1, Bytes.toBytes("a"));
		put.add(ColumnFamilyName, QName2, Bytes.toBytes("b"));

		boolean result = false;

		result = table.checkAndPut(rowKey, ColumnFamilyName, QName2,
				Bytes.toBytes("b"), put);
		// check fail, put fail.
		Assert.assertFalse(result);

		result = table.checkAndPut(rowKey, ColumnFamilyName, QName2, null, put);
		// check ok, put ok.
		Assert.assertTrue(result);

		result = table.checkAndPut(rowKey, ColumnFamilyName, QName2, null, put);
		// check fail, put fail.
		Assert.assertFalse(result);

		result = table.checkAndPut(rowKey, ColumnFamilyName, QName2,
				Bytes.toBytes("b"), put);
		// check ok, put ok.
		Assert.assertTrue(result);
	}

	@Test
	public void testPutAndGet() throws Exception {
		byte[] rowKey = Bytes.toBytes("allen_test_row");
		Put put = new Put(rowKey);
		put.add(ColumnFamilyName, QName1, Bytes.toBytes("a"));
		put.add(ColumnFamilyName, QName3, null);
		table.put(put);

		Get get = new Get(rowKey);
		Result result = table.get(get);

		byte[] q1 = result.getValue(ColumnFamilyName, QName1);
		byte[] q2 = result.getValue(ColumnFamilyName, QName2);
		byte[] q3 = result.getValue(ColumnFamilyName, QName3);

		Assert.assertEquals("a", Bytes.toString(q1));
		// we get null byte array here.
		Assert.assertEquals(null, Bytes.toString(q2));
		// we get empty byte array here. not a null.
		Assert.assertEquals("", Bytes.toString(q3));

		// get a row doesn't exist.
		byte[] rowKey2 = Bytes.toBytes("allen_test_row_not_exist");
		get = new Get(rowKey2);
		result = table.get(get);
		Assert.assertTrue(result.isEmpty());
	}

	@Test
	public void testPutWithoutColumn() throws Exception {
		byte[] rowKey = Bytes.toBytes("allen_test_row");
		Put put = new Put(rowKey);
		try {
			table.put(put);
			Assert.fail();
		} catch (IllegalArgumentException e) {
			// ignore.
		}
	}
}