HBase Java编程示例
HelloWorld.zip 点击(此处)折叠或打开 package elementary; import java.io.IOException; import java.text.SimpleDateFormat; import java.util.ArrayList; import java.util.Date; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; 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.MasterNotRunningException; import org.apache.hadoop.hbase.TableName; 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.Admin; import org.apache.hadoop.hbase.client.BufferedMutator; import org.apache.hadoop.hbase.client.BufferedMutatorParams; import org.apache.hadoop.hbase.client.Connection; import org.apache.hadoop.hbase.client.ConnectionFactory; import org.apache.hadoop.hbase.client.Table; 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.apache.hadoop.util.ThreadUtil; public class HelloWorld { private static Configuration conf = null; private static Connection conn = null; private static Admin admin = null; public static AtomicInteger count = new AtomicInteger(); /** * 初始化配置 */ static { conf = HBaseConfiguration.create(); //如果沒有配置文件，一定要記得手動宣告 conf.set("hbase.zookeeper.quorum", "10.148.137.143"); conf.set("hbase.zookeeper.property.clientPort", "2181"); } static { try { conn = ConnectionFactory.createConnection(); admin = conn.getAdmin(); } catch (IOException e) { e.printStackTrace(); } } static public class MyThread extends Thread { int _start; String _tablename; Connection conn; //BufferedMutator table; Table table; public MyThread(int start, String tablename) { _start = start; _tablename = tablename; } public void run() { String tablename = _tablename; Thread current = Thread.currentThread(); long thread_id = current.getId(); System.out.printf("thread[%d] run\n", thread_id); try { conn = ConnectionFactory.createConnection(); //BufferedMutatorParams params = new BufferedMutatorParams(TableName.valueOf(tablename)); //params.writeBufferSize(1024 * 4); //table = conn.getBufferedMutator(params); table = conn.getTable(TableName.valueOf(tablename)); for (int j=_start; j100; ++j) { for (int i=0; i10000000; ++i) { // zkb_0_0 String zkb = "zkb_" + String.valueOf(_start) + "_" + String.valueOf(i); Put put = new Put(Bytes.toBytes(zkb)); put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field1"),Bytes.toBytes(String.valueOf(i+0))); put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field2"),Bytes.toBytes(String.valueOf(i+1))); put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field3"),Bytes.toBytes(String.valueOf(i+2))); put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field4"),Bytes.toBytes(String.valueOf(i+3))); put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field5"),Bytes.toBytes(String.valueOf(i+4))); put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field6"),Bytes.toBytes(String.valueOf(i+5))); put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field7"),Bytes.toBytes(String.valueOf(i+6))); put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field8"),Bytes.toBytes(String.valueOf(i+7))); put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field9"),Bytes.toBytes(String.valueOf(i+8))); put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field10"),Bytes.toBytes(String.valueOf(i+9))); put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field11"),Bytes.toBytes(String.valueOf(i+10))); put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field12"),Bytes.toBytes(String.valueOf(i+11))); put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field13"),Bytes.toBytes(String.valueOf(i+12))); put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field14"),Bytes.toBytes(String.valueOf(i+13))); put.addColumn(Bytes.toBytes("grade"),Bytes.toBytes("field15"),Bytes.toBytes(String.valueOf(i+14))); //table.mutate(put); table.put(put); int m = HelloWorld.count.incrementAndGet(); if (m % 10000 == 0) { Date dt = new Date(); SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss aa"); String now = sdf.format(dt); System.out.printf("[%s] thread[%d] m=%d, j=%d, i=%d\n", now, thread_id, m, j, i); } } } System.out.printf("thread[%d] over\n", thread_id); } catch (Exception e) { e.printStackTrace(); } } } /** * 建立表格 * @param tablename * @param cfs */ public static void createTable(String tablename, String[] cfs){ try { if (admin.tableExists(TableName.valueOf(tablename))) { System.out.println("table already exists!"); } else { HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(tablename)); for (int i = 0; i cfs.length; i++) { HColumnDescriptor desc = new HColumnDescriptor(cfs[i]); desc.setMaxVersions(3650); tableDesc.addFamily(desc); } byte[][] splitKeys = new byte[][] { Bytes.toBytes("zkb_0_0"), Bytes.toBytes("zkb_10_0"), Bytes.toBytes("zkb_20_0"), Bytes.toBytes("zkb_30_0"), Bytes.toBytes("zkb_40_0"), Bytes.toBytes("zkb_50_0"), Bytes.toBytes("zkb_60_0"), Bytes.toBytes("zkb_70_0"), Bytes.toBytes("zkb_80_0"), Bytes.toBytes("zkb_90_0"), Bytes.toBytes("zkb_100_0") }; admin.createTable(tableDesc, splitKeys); admin.close(); System.out.println("create table " + tablename + " ok."); } } catch (MasterNotRunningException e) { e.printStackTrace(); } catch (ZooKeeperConnectionException e) { e.printStackTrace(); } catch (IOException e) { e.printStackTrace(); } } /** * 刪除表格 * @param tablename */ public static void deleteTable(String tablename){ try { //Connection conn = ConnectionFactory.createConnection(); //Admin admin = conn.getAdmin(); admin.disableTable(TableName.valueOf(tablename)); admin.deleteTable(TableName.valueOf(tablename)); System.out.println("delete table " + tablename + " ok."); } catch (IOException e) { e.printStackTrace(); } } /** * 刪除一筆資料 * @param tableName * @param rowKey */ public static void delRecord (String tableName, String rowKey){ try { Table table = conn.getTable(TableName.valueOf(tableName)); ListDelete> list = new ArrayListDelete>(); Delete del = new Delete(rowKey.getBytes()); list.add(del); table.delete(list); System.out.println("del recored " + rowKey + " ok."); } catch (IOException e) { e.printStackTrace(); } } /** * 取得一筆資料 * @param tableName * @param rowKey */ public static void getOneRecord (String tableName, String rowKey){ try { Table table = conn.getTable(TableName.valueOf(tableName)); Get get = new Get(rowKey.getBytes()); Result rs = table.get(get); ListCell> list = rs.listCells(); for(Cell cell:list){ System.out.print(new String(cell.getRowArray(),cell.getRowOffset(),cell.getRowLength()) + " " ); System.out.print(new String(cell.getFamilyArray(),cell.getFamilyOffset(),cell.getFamilyLength()) + ":" ); System.out.print(new String(cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength()) + " " ); System.out.print(cell.getTimestamp() + " " ); System.out.print(new String(cell.getValueArray(),cell.getValueOffset(),cell.getValueLength()) + " " ); System.out.println(""); } } catch (IOException e) { e.printStackTrace(); } } /** * 取得所有資料 * @param tableName */ public static void getAllRecord (String tableName) { try{ //Connection conn = ConnectionFactory.createConnection(); Table table = conn.getTable(TableName.valueOf(tableName)); Scan scan = new Scan(); ResultScanner resultscanner = table.getScanner(scan); for(Result rs:resultscanner){ ListCell> list = rs.listCells(); for(Cell cell:list){ System.out.print(new String(cell.getRowArray(),cell.getRowOffset(),cell.getRowLength()) + " " ); System.out.print(new String(cell.getFamilyArray(),cell.getFamilyOffset(),cell.getFamilyLength()) + ":" ); System.out.print(new String(cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength()) + " " ); System.out.print(cell.getTimestamp() + " " ); System.out.print(new String(cell.getValueArray(),cell.getValueOffset(),cell.getValueLength()) + " " ); System.out.println(""); } } } catch (IOException e){ e.printStackTrace(); } } /** * 取得Family清單 * @param tableName * @return */ public static ArrayListString> getAllFamilyName(String tableName) { ArrayListString> familyname_list = new ArrayListString>(); try{ //Connection conn = ConnectionFactory.createConnection(); Table table = conn.getTable(TableName.valueOf(tableName)); HTableDescriptor htabledescriptor = table.getTableDescriptor(); HColumnDescriptor[] hdlist = htabledescriptor.getColumnFamilies(); for(int i=0;ihdlist.length;i++){ HColumnDescriptor hd = hdlist[i]; familyname_list.add(hd.getNameAsString()); } } catch (IOException e){ e.printStackTrace(); } return familyname_list; } // java -cp HelloWorld.jar:`ls lib/*.jar|awk '{printf("%s:", $0)}'` elementary.HelloWorld 5 public static void main(String[] args) { System.out.println("HelloWorldX"); if (args.length > 0) System.out.println(args[0]); int start = 0; if (args.length > 1) start = Integer.valueOf(args[1]); if (start 0) start = 0; int num_threads = 16; if (args.length > 2) num_threads = Integer.valueOf(args[2]); try { String tablename = "scores"; String[] familys = {"grade", "course"}; HelloWorld.createTable(tablename, familys); //ExecutorService thread_pool = Executors.newSingleThreadExecutor(); ExecutorService thread_pool = Executors.newFixedThreadPool(num_threads); Thread[] pool = new HelloWorld.MyThread[80]; for (int i=0; ipool.length; ++i) { pool[i] = new HelloWorld.MyThread(i, tablename); thread_pool.execute(pool[i]); } thread_pool.shutdown(); System.out.println("over"); } catch (Exception e) { e.printStackTrace(); } } }
