From 0b99b33494c8952098cbc360455f93cb6f4f63f1 Mon Sep 17 00:00:00 2001 From: Chisu Yu Date: Thu, 20 Sep 2012 02:09:19 +0000 Subject: [PATCH] =?UTF-8?q?[=EC=9C=A0=EC=B9=98=EC=88=98]=20[NOBTS]=20add?= =?UTF-8?q?=20hbase=20client?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit git-svn-id: http://svn.bds.nhncorp.com/pe/hippo-commons/trunk@667 84d0f5b1-2673-498c-a247-62c4ff18d310 --- pom.xml | 11 +- .../profiler/common/hbase/HBaseClient.java | 408 ++++++++++++++++++ .../com/profiler/common/hbase/HBaseQuery.java | 124 ++++++ .../com/profiler/common/hbase/HBaseTypes.java | 16 + .../common/hbase/HBaseClientTest.java | 87 ++++ 5 files changed, 645 insertions(+), 1 deletion(-) create mode 100644 src/main/java/com/profiler/common/hbase/HBaseClient.java create mode 100644 src/main/java/com/profiler/common/hbase/HBaseQuery.java create mode 100644 src/main/java/com/profiler/common/hbase/HBaseTypes.java create mode 100644 src/test/java/com/profiler/common/hbase/HBaseClientTest.java diff --git a/pom.xml b/pom.xml index 5df830a1f..5a2bc0164 100644 --- a/pom.xml +++ b/pom.xml @@ -52,11 +52,20 @@ + + org.apache.hadoop + hadoop-core + 1.0.1 + org.apache.thrift libthrift 0.8.0 - compile + + + org.apache.hbase + hbase + 0.92.1 org.slf4j diff --git a/src/main/java/com/profiler/common/hbase/HBaseClient.java b/src/main/java/com/profiler/common/hbase/HBaseClient.java new file mode 100644 index 000000000..0ac43cb5b --- /dev/null +++ b/src/main/java/com/profiler/common/hbase/HBaseClient.java @@ -0,0 +1,408 @@ +package com.profiler.common.hbase; + +import java.io.IOException; +import java.io.InputStream; +import java.util.ArrayList; +import java.util.Date; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Properties; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hbase.HBaseConfiguration; +import org.apache.hadoop.hbase.HTableDescriptor; +import org.apache.hadoop.hbase.MasterNotRunningException; +import org.apache.hadoop.hbase.ZooKeeperConnectionException; +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.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.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.profiler.common.hbase.HBaseQuery.HbaseColumn; + +public class HBaseClient { + + private static final Logger LOG = LoggerFactory.getLogger(HBaseClient.class); + + public final static String HBASE_ROW_ID = "ROW_KEY"; + + private HTablePool tablePool; + private HBaseAdmin admin; + + private final Map htableList = new HashMap(); + private final Map fieldNameVsType = new HashMap(); + private boolean convertType = false; + + private static class SingletonHolder { + // TODO: configuration this. + public static final HBaseClient INSTANCE = new HBaseClient(10); + } + + public static HBaseClient getInstance() { + return SingletonHolder.INSTANCE; + } + + private HBaseClient(int poolSize) { + Properties properties = readProperties(); + String host = properties.getProperty("hbase.client.host"); + String port = properties.getProperty("hbase.client.port"); + init(host, port, poolSize); + } + + public HBaseClient(String zk, String port, int poolSize) { + init(zk, port, poolSize); + } + + Properties readProperties() { + Properties properties = new Properties(); + InputStream stream = HBaseClient.class.getClassLoader().getResourceAsStream("hbase.properties"); + if(stream == null) { + throw new RuntimeException("hbase.properties not found"); + } + try { + properties.load(stream); + } catch (IOException e) { + throw new RuntimeException("hbase.properties load fail. " + e.getMessage(), e); + } finally { + if(stream != null) { + try { + stream.close(); + } catch (IOException e) { + // 무시 + } + } + } + return properties; + } + + private void init(String zk, String port, int poolSize) { + Configuration cfg = HBaseConfiguration.create(); + if (zk != null) { + cfg.set("hbase.zookeeper.quorum", zk); + } + if (port != null) { + cfg.set("hbase.zookeeper.property.clientPort", port); + } + + tablePool = new HTablePool(cfg, poolSize); + try { + admin = new HBaseAdmin(cfg); + } catch (MasterNotRunningException e) { + e.printStackTrace(); + } catch (ZooKeeperConnectionException e) { + e.printStackTrace(); + } + } + + public void close() { + for (String htableName : htableList.keySet()) { + try { + tablePool.closeTablePool(htableName); + } catch (IOException e) { + e.printStackTrace(); + } + } + } + + public Iterator> getHBaseData(HBaseQuery query) { + ResultSetIterator r = new ResultSetIterator(query); + return r.getIterator(); + } + + private HTableInterface getHTable(String tableName) { + HTableInterface htable = null; + synchronized (htableList) { + htable = htableList.get(tableName); + if (htable == null) { + htable = tablePool.getTable(tableName); + htableList.put(tableName, htable); + } + } + return htable; + } + + public boolean isTableExists(final String tableName) { + try { + return admin.tableExists(tableName); + } catch (IOException e) { + e.printStackTrace(); + return false; + } + } + + public void dropTable(final String tableName) { + try { + admin.disableTable(tableName); + admin.deleteTable(tableName); + } catch (IOException e) { + e.printStackTrace(); + } + } + + public void createTable(HTableDescriptor td) { + try { + admin.createTable(td); + } catch (IOException e) { + e.printStackTrace(); + } + } + + public void flush(byte[] tablename) { + HTable htable = (HTable) tablePool.getTable(tablename); + try { + htable.flushCommits(); + } catch (IOException e) { + e.printStackTrace(); + } + } + + public void insert(byte[] tablename, Put put) { + HTable htable = (HTable) tablePool.getTable(tablename); + try { + htable.put(put); + } catch (IOException e) { + e.printStackTrace(); + } + } + + public void insert(byte[] tablename, List put) { + HTable htable = (HTable) tablePool.getTable(tablename); + try { + htable.put(put); + } catch (IOException e) { + e.printStackTrace(); + } + } + + public void delete(byte[] tablename, Delete delete) { + HTable htable = (HTable) tablePool.getTable(tablename); + try { + htable.delete(delete); + } catch (IOException e) { + e.printStackTrace(); + } + } + + private class ResultSetIterator { + ResultScanner resultScanner = null; + Iterator resultIterator; + List columns; + Iterator> rSetIterator; + + public ResultSetIterator(HBaseQuery query) { + try { + String startRow = query.getStartRow(); + String stopRow = query.getStopRow(); + String tableName = query.getTableName(); + + System.out.println("startRow=" + startRow); + System.out.println("stopRow=" + stopRow); + System.out.println("tableName=" + tableName); + + columns = query.getColumns(); + HTableInterface htable = getHTable(tableName); + + if (query.isSingleRow()) { + System.out.println("Query single row"); + + Get get = new Get(Bytes.toBytes(startRow)); + + if (columns != null) { + for (HbaseColumn column : columns) { + get.addColumn(column.getFamily().getBytes(), column.getColumnName().getBytes()); + } + } + + Result result = htable.get(get); + List resultList = new ArrayList(1); + resultList.add(result); + resultIterator = resultList.iterator(); + + } else { + System.out.println("Query multiple rows"); + Scan scan = new Scan(); + + if (startRow != null) { + scan.setStartRow(startRow.getBytes()); + } + + if (stopRow != null) { + scan.setStopRow(stopRow.getBytes()); + } + + if (columns != null) { + for (HbaseColumn column : columns) { + scan.addColumn(column.getFamily().getBytes(), column.getColumnName().getBytes()); + System.out.println("add column to scanner " + column); + } + } + + LOG.debug("Executing scanner: " + query); + + System.out.println("executing scanner:" + query); + + long start = System.currentTimeMillis(); + + resultScanner = htable.getScanner(scan); + + System.out.println("result scanner:" + resultScanner); + + Iterator it = resultScanner.iterator(); + while (it.hasNext()) { + System.out.println("R=" + it.next()); + } + + LOG.trace("Time taken for scanner: " + (System.currentTimeMillis() - start)); + + resultIterator = resultScanner.iterator(); + } + } catch (Exception e) { + throw new RuntimeException("UNable to execute SCANNER : " + query, e); + } + + if (!resultIterator.hasNext()) { + rSetIterator = new ArrayList>().iterator(); + return; + } + + rSetIterator = new Iterator>() { + public boolean hasNext() { + return hasnext(); + } + + public Map next() { + return getARow(); + } + + public void remove() { + } + }; + } + + private Iterator> getIterator() { + return rSetIterator; + } + + private void addConvertedType(byte[] value, String colName, Map result) { + Integer type = fieldNameVsType.get(colName); + + if (type == null) { + type = HBaseTypes.STRING; + } + switch (type) { + case HBaseTypes.INTEGER: + // result.put(colName, Bytes.toInt(value)); + result.put(colName, Integer.valueOf(Bytes.toString(value))); + break; + case HBaseTypes.FLOAT: + // result.put(colName, Bytes.toFloat(value)); + result.put(colName, Float.valueOf(Bytes.toString(value))); + break; + case HBaseTypes.LONG: + // result.put(colName, Bytes.toLong(value)); + result.put(colName, Long.valueOf(Bytes.toString(value))); + break; + case HBaseTypes.DOUBLE: + // result.put(colName, Bytes.toDouble(value)); + result.put(colName, Double.valueOf(Bytes.toString(value))); + break; + case HBaseTypes.DATE: + result.put(colName, new Date(Bytes.toLong(value))); + // result.put(colName, new + // Date(Long.valueOf(Bytes.toString(value)))); + break; + case HBaseTypes.BOOLEAN: + // result.put(colName, Bytes.toBoolean(value)); + result.put(colName, Boolean.valueOf(Bytes.toString(value))); + break; + case HBaseTypes.BINARY: + result.put(colName, value); + break; + case HBaseTypes.STRING: + result.put(colName, Bytes.toString(value)); + break; + default: + result.put(colName, Bytes.toString(value)); + break; + } + } + + private Map getARow() { + if (resultIterator == null) + return null; + Result res = resultIterator.next(); + + System.out.println("next=" + res); + + Map result = new HashMap(); + if (!res.isEmpty()) { + byte[] value; + if (columns != null) { + for (HbaseColumn column : columns) { + String colName = column.getColumnName(); + value = res.getValue(column.getFamily().getBytes(), column.getColumnName().getBytes()); + + if (value == null) { + continue; + } + + if (!convertType) { + result.put(colName, Bytes.toString(value)); + continue; + } + + // convert type + addConvertedType(value, colName, result); + } + } + value = res.getRow(); + + addConvertedType(value, HBASE_ROW_ID, result); + } + + return result; + } + + private boolean hasnext() { + if (resultIterator == null) + return false; + try { + if (resultIterator.hasNext()) { + return true; + } else { + close(); + return false; + } + + } catch (Exception e) { + close(); + e.printStackTrace(); + return false; + } + } + + private void close() { + try { + if (resultScanner != null) { + resultScanner.close(); + } + } catch (Exception e) { + e.printStackTrace(); + } finally { + } + } + } + + public static final String CONVERT_TYPE = "convertType"; +} diff --git a/src/main/java/com/profiler/common/hbase/HBaseQuery.java b/src/main/java/com/profiler/common/hbase/HBaseQuery.java new file mode 100644 index 000000000..666db7d3d --- /dev/null +++ b/src/main/java/com/profiler/common/hbase/HBaseQuery.java @@ -0,0 +1,124 @@ +package com.profiler.common.hbase; + +import java.util.ArrayList; +import java.util.List; +import java.util.StringTokenizer; + +public class HBaseQuery { + + private String tableName; + + private String startRow; + + private String stopRow; + + private List columns; + + public static class HbaseColumn { + + String family; + + String columnName; + + public HbaseColumn(String family, String columnName) { + this.family = family; + this.columnName = columnName; + } + + public String getFamily() { + return family; + } + + public String getColumnName() { + return columnName; + } + + @Override + public String toString() { + return "{" + family + "|" + columnName + "}"; + } + + } + + public HBaseQuery() { + super(); + } + + public HBaseQuery(String tableName, String startRow, String stopRow, List columns) { + super(); + this.tableName = tableName; + this.startRow = startRow; + this.stopRow = stopRow; + this.columns = columns; + } + + public void setColumns(String columns) { + StringTokenizer st = new StringTokenizer(columns, ","); + + this.columns = new ArrayList(st.countTokens()); + + // columns + while (st.hasMoreElements()) { + String column = ((String) st.nextElement()).trim(); + int separatorIndex = column.indexOf('|'); + + String family = ""; + String columnName = ""; + + if (separatorIndex > -1) { + family = column.substring(0, separatorIndex); + columnName = column.substring(separatorIndex + 1); + } + + HbaseColumn hbcolumn = new HbaseColumn(family, columnName); + this.columns.add(hbcolumn); + } + } + + public boolean isSingleRow() { + if (startRow == null) { + return false; + } + return startRow.equals(stopRow); + } + + // getter and setter + + public String getTableName() { + return tableName; + } + + public void setTableName(String tableName) { + this.tableName = tableName; + } + + public String getStartRow() { + return startRow; + } + + public void setStartRow(String startRow) { + this.startRow = startRow; + } + + public String getStopRow() { + return stopRow; + } + + public void setStopRow(String stopRow) { + this.stopRow = stopRow; + } + + public List getColumns() { + return columns; + } + + public void setColumns(List columns) { + this.columns = columns; + } + + @Override + public String toString() { + return "[tableName=" + tableName + ", startRow=" + startRow + ", stopRow=" + stopRow + ", columns=" + columns + "]"; + } + +} diff --git a/src/main/java/com/profiler/common/hbase/HBaseTypes.java b/src/main/java/com/profiler/common/hbase/HBaseTypes.java new file mode 100644 index 000000000..22c233e67 --- /dev/null +++ b/src/main/java/com/profiler/common/hbase/HBaseTypes.java @@ -0,0 +1,16 @@ +package com.profiler.common.hbase; + +public class HBaseTypes { + public final static int BINARY = 0; + public final static int BOOLEAN = 1; + public final static int DOUBLE = 2; + public final static int FLOAT = 3; + public final static int INTEGER = 4; + public final static int LONG = 5; + public final static int SHORT = 6; + public final static int STRING = 7; + public final static int DATE = 8; + + public HBaseTypes() { + } +} \ No newline at end of file diff --git a/src/test/java/com/profiler/common/hbase/HBaseClientTest.java b/src/test/java/com/profiler/common/hbase/HBaseClientTest.java new file mode 100644 index 000000000..ae731527f --- /dev/null +++ b/src/test/java/com/profiler/common/hbase/HBaseClientTest.java @@ -0,0 +1,87 @@ +package com.profiler.common.hbase; + +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.Map; + +import junit.framework.Assert; + +import org.apache.hadoop.hbase.HColumnDescriptor; +import org.apache.hadoop.hbase.HTableDescriptor; +import org.apache.hadoop.hbase.client.Put; +import org.apache.hadoop.hbase.util.Bytes; +import org.junit.BeforeClass; +import org.junit.Test; + +import com.profiler.common.hbase.HBaseQuery.HbaseColumn; + +public class HBaseClientTest { + + private static final String TABLE_NAME = "TEST_TABLE"; + private static final String COLUMN_FAMILY = "COLUMN_FAMILY"; + private static final HBaseClient client = HBaseClient.getInstance(); + + @BeforeClass + public static void init() { + if (client.isTableExists(TABLE_NAME)) { + client.dropTable(TABLE_NAME); + } + Assert.assertNotNull(client); + } + + @BeforeClass + public static void destroy() { + if (client.isTableExists(TABLE_NAME)) { + client.dropTable(TABLE_NAME); + } + client.close(); + } + + @Test + public void manageTable() { + client.createTable(new HTableDescriptor(TABLE_NAME)); + + Assert.assertTrue("Table is not exists", client.isTableExists(TABLE_NAME)); + + client.dropTable(TABLE_NAME); + client.isTableExists(TABLE_NAME); + + Assert.assertFalse("Table is not dropped", client.isTableExists(TABLE_NAME)); + } + + @Test + public void insertRow() { + HTableDescriptor desc = new HTableDescriptor(TABLE_NAME); + desc.addFamily(new HColumnDescriptor(COLUMN_FAMILY)); + client.createTable(desc); + Assert.assertTrue("Table is not exists", client.isTableExists(TABLE_NAME)); + + List putList = new ArrayList(); + for (int i = 0; i < 10; i++) { + Put put = new Put(Bytes.toBytes(i + "row")); + put.add(Bytes.toBytes(COLUMN_FAMILY), Bytes.toBytes("qual1"), Bytes.toBytes("val1")); + put.add(Bytes.toBytes(COLUMN_FAMILY), Bytes.toBytes("qual2"), Bytes.toBytes("val2")); + putList.add(put); + } + client.insert(Bytes.toBytes(TABLE_NAME), putList); + + client.flush(Bytes.toBytes(TABLE_NAME)); + + try { + Thread.sleep(10000L); + } catch (InterruptedException e) { + e.printStackTrace(); + } + + List list = new ArrayList(); + list.add(new HbaseColumn(COLUMN_FAMILY, "qual1")); + list.add(new HbaseColumn(COLUMN_FAMILY, "qual2")); + Iterator> result = client.getHBaseData(new HBaseQuery(TABLE_NAME, "0", "9", list)); + + while (result.hasNext()) { + Map next = result.next(); + System.out.println(next); + } + } +}