您现在的位置是：首页 > 文章详情

HBase Java编程示例

日期：2016-03-08点击：680收藏

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();
}
}
}