過濾器查詢 · 大數據

[TOC] # 過濾器查詢過濾器的類型很多，但是可以分為兩大類——比較過濾器，專用過濾器過濾器的作用是在服務端判斷數據是否滿足條件，然后只將滿足條件的數據返回給客戶端； hbase過濾器的比較運算符： ~~~ LESS < LESS_OR_EQUAL <= EQUAL = NOT_EQUAL <> GREATER_OR_EQUAL >= GREATER > NO_OP 排除所有 ~~~ Hbase過濾器的比較器（指定比較機制）： ~~~ BinaryComparator 按字節索引順序比較指定字節數組，采用Bytes.compareTo(byte[]) BinaryPrefixComparator 跟前面相同，只是比較左端的數據是否相同 NullComparator 判斷給定的是否為空 BitComparator 按位比較 RegexStringComparator 提供一個正則的比較器，僅支持 EQUAL 和非EQUAL SubstringComparator 判斷提供的子串是否出現在value中 ~~~ # Hbase的過濾器分類 ## 比較過濾器 **行鍵過濾器RowFilter** ~~~ Filter filter1 = new RowFilter(CompareOp.LESS_OR_EQUAL, new BinaryComparator(Bytes.toBytes("row-22"))); scan.setFilter(filter1); ~~~ **列族過濾器FamilyFilter** ~~~ Filter filter1 = new FamilyFilter(CompareFilter.CompareOp.LESS, new BinaryComparator(Bytes.toBytes("colfam3"))); scan.setFilter(filter1); ~~~ **列過濾器QualifierFilter** ~~~ filter = new QualifierFilter(CompareFilter.CompareOp.LESS_OR_EQUAL, new BinaryComparator(Bytes.toBytes("col-2"))); scan.setFilter(filter1); ~~~ **值過濾器 ValueFilter** ~~~ Filter filter = new ValueFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(".4") ); scan.setFilter(filter1); ~~~ ## 專用過濾器 **單列值過濾器 SingleColumnValueFilter** ----會返回滿足條件的整行 ~~~ SingleColumnValueFilter filter = new SingleColumnValueFilter( Bytes.toBytes("colfam1"), Bytes.toBytes("col-5"), CompareFilter.CompareOp.NOT_EQUAL, new SubstringComparator("val-5")); filter.setFilterIfMissing(true); //如果不設置為true，則那些不包含指定column的行也會返回 scan.setFilter(filter1); ~~~ **SingleColumnValueExcludeFilter** 與上相反 **前綴過濾器 PrefixFilter----針對行鍵** ~~~ Filter filter = new PrefixFilter(Bytes.toBytes("row1")); scan.setFilter(filter1); ~~~ **列前綴過濾器 ColumnPrefixFilter** ~~~ Filter filter = new ColumnPrefixFilter(Bytes.toBytes("qual2")); scan.setFilter(filter1); ~~~ ## 過濾器代碼 ~~~ package com.hbase; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.CellScanner; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.*; import org.apache.hadoop.hbase.filter.*; import org.apache.hadoop.hbase.util.Bytes; import org.junit.Before; import org.junit.Test; import java.io.IOException; import java.util.Iterator; public class HbaseDemo { private Configuration conf = null; private Connection conn = null; @Before public void init() throws IOException { //構建個配置 conf = HBaseConfiguration.create(); //對于hbase的客戶端來說,只需要知道hbase所使用的zookeeper集群就可以了 //因為hbase的客戶端找hbase讀寫數據完全不用經過hmaster conf.set("hbase.zookeeper.quorum", "master:2181,slave:2181"); conn = ConnectionFactory.createConnection(conf); } @Test public void testFilter() throws IOException { //針對行鍵的前綴過濾器,row key,前綴過濾 Filter pf = new PrefixFilter(Bytes.toBytes("liu")); testScan(pf); //行過濾器 //比較運算符 //小于,BinaryComparator比較器按照字節字典,排在user002他之前的row key都出來 RowFilter rf1 = new RowFilter(CompareFilter.CompareOp.LESS, new BinaryComparator(Bytes.toBytes("user002"))); //在row key中包含00就符合 RowFilter rf2 = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator("00")); testScan(rf1); System.out.println("*****************"); testScan(rf2); //針對指定一個列的value來過濾,會顯示一個完整的列 //列族名base_info,列標識符password,值是123456 SingleColumnValueFilter scvf = new SingleColumnValueFilter("base_info".getBytes(), "password".getBytes(), CompareFilter.CompareOp.GREATER, "123456".getBytes()); //如果指定的列缺失,則也過濾掉 scvf.setFilterIfMissing(true); testScan(scvf); System.out.println("************"); //正則比較器 //包含以zhang這個字符串開頭的value值,符合這個要求的列 RegexStringComparator comparator1 = new RegexStringComparator("^zhang"); //子串包含si的值,符合這個條件的列 ByteArrayComparable comparator2 = new SubstringComparator("si"); //第三個參數可更換 SingleColumnValueFilter scvf1 = new SingleColumnValueFilter("base_info".getBytes(), "username".getBytes(), CompareFilter.CompareOp.EQUAL, comparator2); testScan(scvf1); //針對列族名的過濾器,返回結果中只會包含滿足條件的列族中的數據 //等于,列族中名稱inf FamilyFilter ff1 = new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("inf"))); //包含這個base前置的列族的對應列 FamilyFilter ff2 = new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryPrefixComparator(Bytes.toBytes("base"))); testScan(ff2); //針對列名的過濾器,返回結果中只會包含滿足條件的列的數據 QualifierFilter qf = new QualifierFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("password"))); QualifierFilter qf2 = new QualifierFilter(CompareFilter.CompareOp.EQUAL, new BinaryPrefixComparator(Bytes.toBytes("us"))); testScan(qf2); //跟SingleColumnValueFilter結果不同，只返回符合條件的該column //前綴過濾 ColumnPrefixFilter cf = new ColumnPrefixFilter("passw".getBytes()); testScan(cf); //指定多個列條件,但是這些條件的或的關系 byte[][] prefixes = {Bytes.toBytes("username"), Bytes.toBytes("password")}; MultipleColumnPrefixFilter mcf = new MultipleColumnPrefixFilter(prefixes); testScan(mcf); //多個過濾器 //等于,前置比較器 FamilyFilter ff20 = new FamilyFilter(CompareFilter.CompareOp.EQUAL, new BinaryPrefixComparator(Bytes.toBytes("base"))); //列前綴比較器 ColumnPrefixFilter cf1 = new ColumnPrefixFilter("passw".getBytes()); //多個過濾器都要滿足 FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL); filterList.addFilter(ff20); filterList.addFilter(cf1); testScan(filterList); } public void testScan(Filter filter) throws IOException { Table t_user_info = conn.getTable(TableName.valueOf("t_user_info")); Scan scan = new Scan(); scan.setFilter(filter); ResultScanner scanner = t_user_info.getScanner(scan); //迭代器 Iterator<Result> iter = scanner.iterator(); while (iter.hasNext()) { //獲取一行記錄 Result result = iter.next(); //獲取到每一個cell CellScanner cellScanner = result.cellScanner(); //遍歷cell while (cellScanner.advance()) { Cell current = cellScanner.current(); byte[] familyArray = current.getFamilyArray(); byte[] valueArray = current.getValueArray(); byte[] qualifierArray = current.getQualifierArray(); byte[] rowArray = current.getRowArray(); System.out.print(new String(rowArray, current.getRowOffset(), current.getRowLength()) + " "); System.out.print(new String(familyArray, current.getFamilyOffset(), current.getFamilyLength())); System.out.print(":" + new String(qualifierArray, current.getQualifierOffset(), current.getQualifierLength())); System.out.print(" " + new String(valueArray, current.getValueOffset(), current.getValueLength())); System.out.println(); } System.out.println("-----------------------------"); } } } ~~~ # 分頁過濾器 PageFilter ~~~ package com.hbase; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.CellScanner; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.*; import org.apache.hadoop.hbase.filter.*; import org.apache.hadoop.hbase.util.Bytes; import org.junit.Before; import org.junit.Test; import java.io.IOException; import java.util.Iterator; public class HbaseDemo { private Configuration conf = null; private Connection conn = null; @Before public void init() throws IOException { //構建個配置 conf = HBaseConfiguration.create(); //對于hbase的客戶端來說,只需要知道hbase所使用的zookeeper集群就可以了 //因為hbase的客戶端找hbase讀寫數據完全不用經過hmaster conf.set("hbase.zookeeper.quorum", "master:2181,slave:2181"); conn = ConnectionFactory.createConnection(conf); } //分頁查詢 @Test public void pageScan() throws IOException, InterruptedException { final byte[] POSTFIX = {0x00}; //獲取表 Table table = conn.getTable(TableName.valueOf("t_user_info")); //分頁過濾器 PageFilter filter = new PageFilter(3); //起始行號 byte[] lastRow = null; //總共的記錄 int totalRows = 0; while (true) { Scan scan = new Scan(); scan.setFilter(filter); //當上次起始行不為空 if (lastRow != null) { //設置本次查詢的起始行鍵 //上次起始行加上后置,加后置可以獲取上次結束行作為本次的起始行 byte[] startRow = Bytes.add(lastRow, POSTFIX); //設置為起始行 scan.setStartRow(startRow); } //獲取整個掃描的結果 ResultScanner scanner = table.getScanner(scan); //定義本地的行號 int localRows = 0; //結果 Result result; //遍歷一頁的結果 while ((result = scanner.next()) != null) { //localRows顯示本地行號每頁中的行號,result會調用toString System.out.println(++localRows + ":" + result); //全局行號++ totalRows++; //上次起始的行號設置為這次結束的行號 lastRow = result.getRow(); } scanner.close(); if (localRows == 0) { break; } } //打印本次總行數 System.out.println("total rows:" + totalRows); } } ~~~