HBase Java编程示例-低调大师

HBase Java编程示例

2016-03-08 712

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

微信关注我们

原文链接：https://yq.aliyun.com/articles/497317

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

CDH5之Balancer难以在快速增长的集群上平衡大量的数据

背景：公司在线上使用了CDH5集群，一开始由于疏忽，忘记了在计划任务中定期执行Balancer来平衡各节点的数据。后来，在引入大量的Job之后，数据增长非常迅猛，有很多节点开始出现利用率超过99.9%的情况，部分Job甚至开始Failed。于是我们便执行Balancer来清理数据，结果发现有26T的数据需要平衡，而Balancer每次只移动50G的数据，并且耗时30分钟，而集群每个小时新写入的数据会导致又有40-60G的数据需要平衡。这样一来，Balancer就根本无法胜任了。 14/10/14 20:31:11 INFO balancer.Balancer:Need to move 26.49 TB to make the cluster balanced. 14/10/14 20:31:11 INFO balancer.Balancer:Decided to move 10 GB bytes from 10.100.1.10:50010 to 10.100.1.60:50010 14/10/14 20:31:11 INFO balancer.Balancer:Decided ...

2016-03-08

645

Hadoop2.7实战v1.0之start-balancer.sh与hdfs balancer数据均衡

Hadoop2.7实战v1.0之start-balancer.sh与hdfs balancer数据均衡【修正版】适用场景： a.当动态添加或者删除集群的数据节点，必然会使各节点的数据不均衡 b.当正常维护时 1.对hdfs负载设置均衡，因为默认的数据传输带宽比较低，可以设置为64M，即hdfs dfsadmin -setBalancerBandwidth 67108864即可点击(此处)折叠或打开 [root@sht-sgmhadoopnn-01 ~]#cd /hadoop/hadoop-2.7.2/bin [root@sht-sgmhadoopdn-01 bin]#./hdfs dfsadmin-setBalancerBandwidth 67108864 Balancer bandwidth is set to 67108864forsht-sgmhadoopnn-01/172.16.101.55:8020 Balancer bandwidth is set to 67108864forsht-sgmhadoopnn-02/172.16.101.56:8020 2.默认bala...

2016-03-09

672

发表评论

资源下载

更多资源

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。