Hbase學習筆記3之集群操作

數(shù)據(jù)融合

當我們手動刷新數(shù)據(jù)到磁盤的時候，會創(chuàng)建一個新的文件。然后這些大小不一的小文件，最后會進行自動合并。如下所示，我們先進行幾次刷新。

每次刷新都會在列族目錄下多出一個對應的數(shù)據(jù)文件來。

第二個文件：

第三個文件的時候發(fā)現(xiàn)文件進行了合并。

查詢結果為前幾次put進去的值。

Java API

接下來我們使用Java來操作Hbase數(shù)據(jù)庫

初始化

@Before
public void init() throws IOException {
//創(chuàng)建配置文件對象
        conf = HBaseConfiguration.create();
//加載zookeeper的配置
        conf.set("hbase.zookeeper.quorum","node2,node3,node4");
//獲取連接
        conn = ConnectionFactory.createConnection(conf);
//獲取對象
        admin = conn.getAdmin();
//獲取數(shù)據(jù)操作對象
        table = conn.getTable(tableName);
}

創(chuàng)建表

 @Test
public void createTable() throws IOException {
//定義表描述對象
        TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName);
//定義列族描述對象
        ColumnFamilyDescriptorBuilder columnFamilyDescriptorBuilder = ColumnFamilyDescriptorBuilder.newBuilder("cf".getBytes());
//添加列族信息給表
        tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptorBuilder.build());
if(admin.tableExists(tableName)){
//禁用表
            admin.disableTable(tableName);
admin.deleteTable(tableName);
}
//創(chuàng)建表
        admin.createTable(tableDescriptorBuilder.build());
}

新增數(shù)據(jù)

@Test
public void insert() throws IOException {
Put put = new Put(Bytes.toBytes("2222"));
put.addColumn(Bytes.toBytes("cf"),Bytes.toBytes("name"),Bytes.toBytes("lisi"));
put.addColumn(Bytes.toBytes("cf"),Bytes.toBytes("age"),Bytes.toBytes("341"));
put.addColumn(Bytes.toBytes("cf"),Bytes.toBytes("sex"),Bytes.toBytes("women"));
table.put(put);
}

查詢數(shù)據(jù)

 @Test
public void scan() throws IOException {
Scan scan = new Scan();
ResultScanner rss = table.getScanner(scan);
for (Result rs: rss) {
Cell cell1 = rs.getColumnLatestCell(Bytes.toBytes("cf"),Bytes.toBytes("name"));
Cell cell2 = rs.getColumnLatestCell(Bytes.toBytes("cf"),Bytes.toBytes("age"));
Cell cell3 = rs.getColumnLatestCell(Bytes.toBytes("cf"),Bytes.toBytes("sex"));
String name = Bytes.toString(CellUtil.cloneValue(cell1));
String age = Bytes.toString(CellUtil.cloneValue(cell2));
String sex = Bytes.toString(CellUtil.cloneValue(cell3));
System.out.println(name);
System.out.println(age);
System.out.println(sex);
}
}

插入十萬數(shù)據(jù)

/**
     * 假設有10個用戶，每個用戶一年產生10000條記錄
     */
@Test
public void insertMangData() throws Exception {
List<Put> puts = new ArrayList<>();
for(int i = 0;i<10;i++){
String phoneNumber = getNumber("158");
for(int j = 0 ;j<10000;j++){
String dnum = getNumber("177");
String length = String.valueOf(random.nextInt(100));
String date = getDate("2019");
String type = String.valueOf(random.nextInt(2));
//rowkey
                String rowkey = phoneNumber+"_"+(Long.MAX_VALUE-sdf.parse(date).getTime());
Put put = new Put(Bytes.toBytes(rowkey));
put.addColumn(Bytes.toBytes("cf"),Bytes.toBytes("dnum"),Bytes.toBytes(dnum));
put.addColumn(Bytes.toBytes("cf"),Bytes.toBytes("length"),Bytes.toBytes(length));
put.addColumn(Bytes.toBytes("cf"),Bytes.toBytes("date"),Bytes.toBytes(date));
put.addColumn(Bytes.toBytes("cf"),Bytes.toBytes("type"),Bytes.toBytes(type));
puts.add(put);
}
}
table.put(puts);
}

SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");

private String getDate(String s) {
return s+String.format("%02d%02d%02d%02d%02d",random.nextInt(12)+1,random.nextInt(31),random.nextInt(24),random.nextInt(60),random.nextInt(60));
}

Random random = new Random();
public String getNumber(String str){
return str+String.format("%08d",random.nextInt(99999999));
}

查詢指定條件

/**
     * 查詢某一個用戶3月份的通話記錄
     */
@Test
public void scanByCondition() throws Exception {
Scan scan = new Scan();
String startRow = "15883348450_"+(Long.MAX_VALUE-sdf.parse("20190331000000").getTime());
String stopRow = "15883348450_"+(Long.MAX_VALUE-sdf.parse("20190301000000").getTime());
scan.withStartRow(Bytes.toBytes(startRow));
scan.withStopRow(Bytes.toBytes(stopRow));
ResultScanner rss = table.getScanner(scan);
for (Result rs:rss) {
System.out.print(Bytes.toString(CellUtil.cloneValue(rs.getColumnLatestCell(Bytes.toBytes("cf"),Bytes.toBytes("dnum")))));
System.out.print("--"+Bytes.toString(CellUtil.cloneValue(rs.getColumnLatestCell(Bytes.toBytes("cf"),Bytes.toBytes("type")))));
System.out.print("--"+Bytes.toString(CellUtil.cloneValue(rs.getColumnLatestCell(Bytes.toBytes("cf"),Bytes.toBytes("length")))));
System.out.println("--"+Bytes.toString(CellUtil.cloneValue(rs.getColumnLatestCell(Bytes.toBytes("cf"),Bytes.toBytes("date")))));
}
}

/**
     * 查詢某個用戶所有的主叫電話（type=1）
     *  某個用戶
     *  type=1
     *
     */
@Test
public void getType() throws IOException {
Scan scan = new Scan();
//創(chuàng)建過濾器集合
        FilterList filters = new FilterList(FilterList.Operator.MUST_PASS_ALL);
//創(chuàng)建過濾器
        SingleColumnValueFilter filter1 = new SingleColumnValueFilter(Bytes.toBytes("cf"),Bytes.toBytes("type"),CompareOperator.EQUAL,Bytes.toBytes("1"));
filters.addFilter(filter1);
//前綴過濾器
        PrefixFilter filter2 = new PrefixFilter(Bytes.toBytes("15883348450"));
filters.addFilter(filter2);
scan.setFilter(filters);
ResultScanner rss = table.getScanner(scan);
for (Result rs:rss) {
System.out.print(Bytes.toString(CellUtil.cloneValue(rs.getColumnLatestCell(Bytes.toBytes("cf"),Bytes.toBytes("dnum")))));
System.out.print("--"+Bytes.toString(CellUtil.cloneValue(rs.getColumnLatestCell(Bytes.toBytes("cf"),Bytes.toBytes("type")))));
System.out.print("--"+Bytes.toString(CellUtil.cloneValue(rs.getColumnLatestCell(Bytes.toBytes("cf"),Bytes.toBytes("length")))));
System.out.println("--"+Bytes.toString(CellUtil.cloneValue(rs.getColumnLatestCell(Bytes.toBytes("cf"),Bytes.toBytes("date")))));
}
}