HBase Java编程示例
点击(此处)折叠或打开
- 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();
- }
- }
-
- }
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道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 ...
- 下一篇
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...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
推荐阅读
最新文章
- CentOS7安装Docker,走上虚拟化容器引擎之路
- Docker使用Oracle官方镜像安装(12C,18C,19C)
- CentOS8编译安装MySQL8.0.19
- Docker安装Oracle12C,快速搭建Oracle学习环境
- CentOS7编译安装Gcc9.2.0,解决mysql等软件编译问题
- CentOS6,7,8上安装Nginx,支持https2.0的开启
- CentOS关闭SELinux安全模块
- CentOS7设置SWAP分区,小内存服务器的救世主
- CentOS7编译安装Cmake3.16.3,解决mysql等软件编译问题
- CentOS8安装Docker,最新的服务器搭配容器使用