hbase的api使用：scan,get,put,delete

╰半橙微兮° 2022-04-22 00:36 272阅读 0赞

### 文章目录 ###

*  1, 测试hbase api：增，删，查询
 *  2, 优化hbase查询： 批次查询
 *  3, 优化hbase数据插入： 批量插入
 *  4, 数据同步/导入：hbase1到hbase2
 *  5, hbase表统计行数

*  一般化的curd
 *  批次查询，插入  
    在idea中导入maven依赖，创建Junit Test类

<dependency>
                <groupId>org.apache.hbase</groupId>
                <artifactId>hbase-client</artifactId>
                <version>1.2.6</version>
            </dependency>

编写@before方法初始化属性，创建database, table

Configuration conf;
    	Connection conn;
    	Admin admin;
    
    	@Before
    	public void before()throws IOException{ 
    		 conf = HBaseConfiguration.create();
    		 conn = ConnectionFactory.createConnection(conf) ;
    		 admin = conn.getAdmin();
    	}
    	/** * 创建database */
    	@Test
    	public void testCreateNamespace() throws IOException { 
    		//名字空间描述符
    		NamespaceDescriptor nd = NamespaceDescriptor.create("ns2").build();
    		admin.createNamespace(nd);
    		admin.close();
    	}
    	/** * 创建表 */
    	@Test
    	public void testCreatTable() throws IOException { 
    		//创建表明对象
    		TableName tname = TableName.valueOf("ns2:t2") ;
    		HTableDescriptor td = new HTableDescriptor(tname) ;
    		//创建列族
    		HColumnDescriptor f1 = new HColumnDescriptor("f1") ;
    		HColumnDescriptor f2 = new HColumnDescriptor("f2") ;
    
    		//给表添加列族
    		td.addFamily(f1) ;
    		td.addFamily(f2) ;
    
    		admin.createTable(td);
    		admin.close();
    	}

# 1, 测试hbase api：增，删，查询 #

/** * 插入数据 */
    	@Test
    	public void testPut() throws IOException { 
    		Table t = conn.getTable(TableName.valueOf("ns2:t2")) ;
    		Put put = new Put(Bytes.toBytes("row2")) ;
    		put.addColumn(Bytes.toBytes("f1") , Bytes.toBytes("id") , Bytes.toBytes(2)) ;
    		put.addColumn(Bytes.toBytes("f1") , Bytes.toBytes("name") , Bytes.toBytes("tom2")) ;
    		put.addColumn(Bytes.toBytes("f1") , Bytes.toBytes("age") , Bytes.toBytes(11)) ;
    		t.put(put);
    		t.close();
    		conn.close();
    	}
    	/** *删除数据 */
    	@Test
    	public void testDelete() throws IOException { 
    		Table t = conn.getTable(TableName.valueOf("ns2:t2")) ;
    		Delete d = new Delete(Bytes.toBytes("row1")) ;
    		d.addColumn(Bytes.toBytes("f1") , Bytes.toBytes("id")) ;
    		t.delete(d);
    		t.close();
    		conn.close();
    	}
    	/** * 查询： 获取某一行 某列 */
    	@Test
    	public void testGet() throws IOException { 
    		Table t = conn.getTable(TableName.valueOf("ns2:t2")) ;
    		Get get = new Get(Bytes.toBytes("row1")) ;
    // get.addColumn(Bytes.toBytes("f1") , Bytes.toBytes("name")) ;
    		Result r = t.get(get) ;
            List<Cell> cells = r.listCells();
            System.out.println("=================");
            for (Cell cell : cells) { 
                String row = Bytes.toString(CellUtil.cloneRow(cell));
                String f = Bytes.toString(CellUtil.cloneFamily(cell));
                String col = Bytes.toString(CellUtil.cloneQualifier(cell));
                long ver = cell.getTimestamp();
                String val = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.printf("%s/%s:%s/%d=%s\r\n" , row,f,col,ver,val);
            }
    		t.close();
    		conn.close();
    	}
    
    	/** * 查询： 扫描多行 */
    	@Test
    	public void testScan() throws IOException { 
    		Table t = conn.getTable(TableName.valueOf("ns2:t2")) ;
    		Scan scan = new Scan() ;
    		ResultScanner scanner = t.getScanner(scan) ;
    		Iterator<Result> it = scanner.iterator();
    		while(it.hasNext()){ 
                Result r = it.next();
                List<Cell> cells = r.listCells();
                System.out.println("=================");
                for (Cell cell : cells) { 
                    String row = Bytes.toString(CellUtil.cloneRow(cell));
                    String f = Bytes.toString(CellUtil.cloneFamily(cell));
                    String col = Bytes.toString(CellUtil.cloneQualifier(cell));
                    long ver = cell.getTimestamp();
                    String val = Bytes.toString(CellUtil.cloneValue(cell));
                    System.out.printf("%s/%s:%s/%d=%s\r\n" , row,f,col,ver,val);
                }
    		}
    		t.close();
    		conn.close();
    	}

# 2, 优化hbase查询： 批次查询 #

/** * 带缓存查询：scan.setCaching(10000) ; */
    	@Test
    	public void cacheScan() throws IOException { 
    		Table t = conn.getTable(TableName.valueOf("ns2:t2"));
    		Scan scan = new Scan();
    		//切忌全表扫描，设置起始结束行
    		scan.setStartRow(Bytes.toBytes("row000001")) ;
    		scan.setStopRow(Bytes.toBytes("row010001")) ;
    		System.out.println(scan.getCaching());
    		scan.setCaching(10000) ;
    
    		ResultScanner scanner = t.getScanner(scan);
    		Iterator<Result> it = scanner.iterator();
    		long start = System.currentTimeMillis() ;
    		int index = 0 ;
    		while (it.hasNext()) { 
    			it.next().getRow();
    			//System.out.println(index ++);
    		}
    		System.out.println(System.currentTimeMillis() - start);
    		t.close();
    		conn.close();
    	}
    
    	/** * 批次查询： scan.setBatch(10); */
    	@Test
    	public void batchScan() throws IOException { 
    		Table t = conn.getTable(TableName.valueOf("ns2:t3"));
    		Scan scan = new Scan();
    		scan.setBatch(10);
    		Iterator<Result> it = t.getScanner(scan).iterator() ;
    		while(it.hasNext()){ 
                Result r = it.next();
                List<Cell> cells = r.listCells();
                System.out.println("=================");
                for (Cell cell : cells) { 
                    String row = Bytes.toString(CellUtil.cloneRow(cell));
                    String f = Bytes.toString(CellUtil.cloneFamily(cell));
                    String col = Bytes.toString(CellUtil.cloneQualifier(cell));
                    long ver = cell.getTimestamp();
                    String val = Bytes.toString(CellUtil.cloneValue(cell));
                    System.out.printf("%s/%s:%s/%d=%s\r\n" , row,f,col,ver,val);
                }
    		}
    		t.close();
    		conn.close();
    	}
    }

# 3, 优化hbase数据插入： 批量插入 #

/** * 批量插入数据 */
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.hbase.HBaseConfiguration;
    import org.apache.hadoop.hbase.TableName;
    import org.apache.hadoop.hbase.client.*;
    import org.apache.hadoop.hbase.util.Bytes;
    
    import java.io.IOException;
    import java.util.ArrayList;
    import java.util.List;
    
    public class HbaseTest { 
        public static void main(String[] args) throws IOException { 
            Configuration conf = HBaseConfiguration.create();
            Connection conn = ConnectionFactory.createConnection(conf);
            String tableName = "test1";
            //############## 1, BufferedMutator ######## 耗时：1987ms 2080 2083 1968
            mutatePute(conf, conn, tableName);
            //############## 2, 关闭自动刷写&& put(list) ######## 耗时：1984ms 2169 2247 1939
    // listPut(conn, tableName);
        }
    
        private static void listPut(Connection conn, String tableName) throws IOException { 
            ThreadLocal<List<Put>> threadLocal = new ThreadLocal<List<Put>>();
            int CACHE_LIST_SIZE = 500; //批量提交数据条数
            HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
            table.setAutoFlushTo(false);//10万条数据(开启批量写入：1931ms； 默认自动刷写：4715ms)
            table.setWriteBufferSize(8 * 1024 * 1024); //flush前，存储8MB写入缓存
            //多个put以前写入
            long start = System.currentTimeMillis();
            for (int i = 0; i < 100000; i++) { 
                try { 
                    List<Put> list = threadLocal.get();
                    if (list == null) { 
                        list = new ArrayList<Put>();
                    }
                    Put put = new Put(Bytes.toBytes(i + ""));
                    put.add(Bytes.toBytes("f"), Bytes.toBytes("name"), Bytes.toBytes(100000 + i));
                    list.add(put);
    
                    if (list.size() >= CACHE_LIST_SIZE) { 
                        table.put(list);
                        list.clear();
                    } else { 
                        threadLocal.set(list);
                    }
                } catch (IOException e) { 
                    e.printStackTrace();
                }
            }
            //清空list里面的数据
            List<Put> list = threadLocal.get();
            if (list != null && list.size() > 0) { 
                table.put(list);
            }
            table.flushCommits();
            System.out.println(System.currentTimeMillis() - start);
        }
    
        private static void mutatePute(Configuration conf, Connection conn, String tableName) throws IOException { 
            BufferedMutator bufferedMutator = conn.getBufferedMutator(
                    new BufferedMutatorParams(TableName.valueOf(tableName)).pool(HTable.getDefaultExecutor(conf)).writeBufferSize(8 * 1024 * 1024)
            );//flush前，存储8MB写入缓存
            long start = System.currentTimeMillis();
            for (int i = 0; i <= 100000; i++) { 
                Put put = new Put(Bytes.toBytes(i + ""));
                //put.setDurability(Durability.SKIP_WAL) ;
                put.addColumn(Bytes.toBytes("f"), Bytes.toBytes("age"), Bytes.toBytes(i % 100));
                bufferedMutator.mutate(put);
            }
            //刷新并提交
            bufferedMutator.close();
            System.out.println(System.currentTimeMillis() - start);
        }
    }

# 4, 数据同步/导入：hbase1到hbase2 #

/** * 读hbaes1数据，插入hbase2 */
        @Test
        public void testScanPut() throws IOException { 
            //hbase1: 查询全表数据
            //Configuration conf = HBaseConfiguration.create();
            Configuration conf = new Configuration();
            conf.setClassLoader(HBaseConfiguration.class.getClassLoader());
            conf.addResource("hbase-site-1.xml");
            Connection conn = ConnectionFactory.createConnection(conf) ;
            Table t = conn.getTable(TableName.valueOf("t2")) ;
            Scan scan = new Scan() ;
            scan.setCaching(10000) ;
            ResultScanner scanner = t.getScanner(scan) ;
            Iterator<Result> it = scanner.iterator();
    
            //hbase2: 写入hbase数据
            Configuration conf2 = new Configuration();
            conf2.setClassLoader(HBaseConfiguration.class.getClassLoader());
            conf2.addResource("hbase-site-2.xml");
            Connection conn2 = ConnectionFactory.createConnection(conf2) ;
            BufferedMutator bufferedMutator = conn2.getBufferedMutator(
                    new BufferedMutatorParams(TableName.valueOf("t1"))
                            .pool(HTable.getDefaultExecutor(conf2))
                            .writeBufferSize(8*1024*1024));//8MB
    
    
            while(it.hasNext()){ 
                Result r = it.next();
                List<Cell> cells = r.listCells();
                System.out.println("=================");
                for (Cell cell : cells) { 
                    String row = Bytes.toString(CellUtil.cloneRow(cell));
                    String f = Bytes.toString(CellUtil.cloneFamily(cell));
                    String col = Bytes.toString(CellUtil.cloneQualifier(cell));
                    long ver = cell.getTimestamp();
                    String val = Bytes.toString(CellUtil.cloneValue(cell));
                    System.out.printf("%s/%s:%s/%d=%s\r\n" , row,f,col,ver,val);
    
                    //
                    Put put2 = new Put(Bytes.toBytes(row)) ;
                    put2.addColumn(Bytes.toBytes("f1") , Bytes.toBytes(col) , Bytes.toBytes(val)) ;
                    bufferedMutator.mutate(put2);
                }
            }
            bufferedMutator.close();
            conn2.close();
            //
            t.close();
            conn.close();
        }

# 5, hbase表统计行数 #

*  表启用协处理器：[https://blogs.apache.org/hbase/entry/coprocessor\_introduction][https_blogs.apache.org_hbase_entry_coprocessor_introduction]
 *  `alter 'www', METHOD => 'table_att', 'coprocessor' =>'|org.apache.hadoop.hbase.coprocessor.AggregateImplementation||'`

![在这里插入图片描述][20210708172511274.png]

import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.hbase.TableName;
    import org.apache.hadoop.hbase.client.Scan;
    import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
    import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter;
    
    public class Test2 { 
        public static void main(String[] args) throws Throwable { 
            Configuration conf = new Configuration();
            conf.setLong("hbase.rpc.timeout",60000);
            conf.setLong("hbase.client.scanner.caching", 1000);
            conf.set("hbase.zookeeper.quorum","hadoop01:2181");
            //conf.set("hbase.rootdir", "hdfs://windows:8020/hbase");
            AggregationClient client= new AggregationClient(conf);
            Scan scan = new Scan();
            long l1 = System.currentTimeMillis();
            //scan.addFamily("f".getBytes());
            //表启用协处理器：https://blogs.apache.org/hbase/entry/coprocessor_introduction
            // alter 'www', METHOD => 'table_att', 'coprocessor' =>'|org.apache.hadoop.hbase.coprocessor.AggregateImplementation||'
            //TableName table = TableName.valueOf("www");
            TableName table = TableName.valueOf("KYLIN_DQP636J5PA");
            long l = client.rowCount(table, new LongColumnInterpreter(), scan);
            System.out.println(l);
            System.out.println(System.currentTimeMillis()-l1);
        }
    }

[https_blogs.apache.org_hbase_entry_coprocessor_introduction]: https://blogs.apache.org/hbase/entry/coprocessor_introduction
[20210708172511274.png]: /images/20220422/68fab7a5955a4781af755bee327ea135.png