为了账号安全,请及时绑定邮箱和手机立即绑定

请教hbase如何用过滤器实现项目某个求总数量的统计?

/ 猿问

请教hbase如何用过滤器实现项目某个求总数量的统计?

BIG阳 2019-10-15 14:10:24

hbase如何用过滤器实现项目某个求总数量的统计


查看完整描述

4 回答

?
米脂

HBase为筛选数据提供了一组过滤器,通过这个过滤器可以在HBase中的数据的多个维度(行,列,数据版本)上进行对数据的筛选操作,也就是说过滤器最终能够筛选的数据能够细化到具体的一个存储单元格上(由行键,列明,时间戳定位)。通常来说,通过行键,值来筛选数据的应用场景较多。

1. RowFilter:筛选出匹配的所有的行,对于这个过滤器的应用场景,是非常直观的:使用BinaryComparator可以筛选出具有某个行键的行,或者通过改变比较运算符(下面的例子中是CompareFilter.CompareOp.EQUAL)来筛选出符合某一条件的多条数据,以下就是筛选出行键为row1的一行数据:

[java] view plaincopy
Filter rf = new RowFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("row1"))); // OK 筛选出匹配的所有的行

2. PrefixFilter:筛选出具有特定前缀的行键的数据。这个过滤器所实现的功能其实也可以由RowFilter结合RegexComparator来实现,不过这里提供了一种简便的使用方法,以下过滤器就是筛选出行键以row为前缀的所有的行:
[java] view plaincopy在CODE上查看代码片派生到我的代码片
Filter pf = new PrefixFilter(Bytes.toBytes("row")); // OK 筛选匹配行键的前缀成功的行

3. KeyOnlyFilter:这个过滤器唯一的功能就是只返回每行的行键,值全部为空,这对于只关注于行键的应用场景来说非常合适,这样忽略掉其值就可以减少传递到客户端的数据量,能起到一定的优化作用:
[java] view plaincopy在CODE上查看代码片派生到我的代码片
Filter kof = new KeyOnlyFilter(); // OK 返回所有的行,但值全是空

4. RandomRowFilter:从名字上就可以看出其大概的用法,本过滤器的作用就是按照一定的几率(<=0会过滤掉所有的行,>=1会包含所有的行)来返回随机的结果集,对于同样的数据集,多次使用同一个RandomRowFilter会返回不通的结果集,对于需要随机抽取一部分数据的应用场景,可以使用此过滤器:
[java] view plaincopy在CODE上查看代码片派生到我的代码片
Filter rrf = new RandomRowFilter((float) 0.8); // OK 随机选出一部分的行

5. InclusiveStopFilter:扫描的时候,我们可以设置一个开始行键和一个终止行键,默认情况下,这个行键的返回是前闭后开区间,即包含起始行,单不包含中指行,如果我们想要同时包含起始行和终止行,那么我们可以使用此过滤器:
[java] view plaincopy在CODE上查看代码片派生到我的代码片
Filter isf = new InclusiveStopFilter(Bytes.toBytes("row1")); // OK 包含了扫描的上限在结果之内

6. FirstKeyOnlyFilter:如果你只想返回的结果集中只包含第一列的数据,那么这个过滤器能够满足你的要求。它在找到每行的第一列之后会停止扫描,从而使扫描的性能也得到了一定的提升:
[java] view plaincopy在CODE上查看代码片派生到我的代码片
Filter fkof = new FirstKeyOnlyFilter(); // OK 筛选出第一个每个第一个单元格

7. ColumnPrefixFilter:顾名思义,它是按照列名的前缀来筛选单元格的,如果我们想要对返回的列的前缀加以限制的话,可以使用这个过滤器:
[java] view plaincopy在CODE上查看代码片派生到我的代码片
Filter cpf = new ColumnPrefixFilter(Bytes.toBytes("qual1")); // OK 筛选出前缀匹配的列

8. ValueFilter:按照具体的值来筛选单元格的过滤器,这会把一行中值不能满足的单元格过滤掉,如下面的构造器,对于每一行的一个列,如果其对应的值不包含ROW2_QUAL1,那么这个列就不会返回给客户端:
[java] view plaincopy在CODE上查看代码片派生到我的代码片
Filter vf = new ValueFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator("ROW2_QUAL1")); // OK 筛选某个(值的条件满足的)特定的单元格

9. ColumnCountGetFilter:这个过滤器来返回每行最多返回多少列,并在遇到一行的列数超过我们所设置的限制值的时候,结束扫描操作:
[java] view plaincopy在CODE上查看代码片派生到我的代码片
Filter ccf = new ColumnCountGetFilter(2); // OK 如果突然发现一行中的列数超过设定的最大值时,整个扫描操作会停止

10. SingleColumnValueFilter:用一列的值决定这一行的数据是否被过滤。在它的具体对象上,可以调用setFilterIfMissing(true)或者setFilterIfMissing(false),默认的值是false,其作用是,对于咱们要使用作为条件的列,如果这一列本身就不存在,那么如果为true,这样的行将会被过滤掉,如果为false,这样的行会包含在结果集中。

[java] view plaincopy在CODE上查看代码片派生到我的代码片
SingleColumnValueFilter scvf = new SingleColumnValueFilter(
Bytes.toBytes("colfam1"),
Bytes.toBytes("qual2"),
CompareFilter.CompareOp.NOT_EQUAL,
new SubstringComparator("BOGUS"));
scvf.setFilterIfMissing(false);
scvf.setLatestVersionOnly(true); // OK

11. SingleColumnValueExcludeFilter:这个与10种的过滤器唯一的区别就是,作为筛选条件的列的不会包含在返回的结果中。
12. SkipFilter:这是一种附加过滤器,其与ValueFilter结合使用,如果发现一行中的某一列不符合条件,那么整行就会被过滤掉:

[java] view plaincopy在CODE上查看代码片派生到我的代码片
Filter skf = new SkipFilter(vf); // OK 发现某一行中的一列需要过滤时,整个行就会被过滤掉

13. WhileMatchFilter:这个过滤器的应用场景也很简单,如果你想要在遇到某种条件数据之前的数据时,就可以使用这个过滤器;当遇到不符合设定条件的数据的时候,整个扫描也就结束了:
[java] view plaincopy在CODE上查看代码片派生到我的代码片
Filter wmf = new WhileMatchFilter(rf); // OK 类似于Python itertools中的takewhile

14. FilterList:用于综合使用多个过滤器。其有两种关系:FilterList.Operator.MUST_PASS_ONE和FilterList.Operator.MUST_PASS_ALL,默认的是FilterList.Operator.MUST_PASS_ALL,顾名思义,它们分别是AND和OR的关系,并且FilterList可以嵌套使用FilterList,使我们能够表达更多的需求:

[java] view plaincopy在CODE上查看代码片派生到我的代码片
List<Filter> filters = new ArrayList<Filter>();
filters.add(rf);
filters.add(vf);
FilterList fl = new FilterList(FilterList.Operator.MUST_PASS_ALL, filters); // OK 综合使用多个过滤器, AND 和 OR 两种关系

以上,是对于HBase内置的过滤器的部分总结,以下代码是数据写入代码:

[java] view plaincopy在CODE上查看代码片派生到我的代码片
package com.reyun.hbase;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

public class HBaseDataFeeding {
private final static byte[] ROW1 = Bytes.toBytes("row1");
private final static byte[] ROW2 = Bytes.toBytes("row2");
private final static byte[] COLFAM1 = Bytes.toBytes("colfam1");
private final static byte[] COLFAM2 = Bytes.toBytes("colfam2");
private final static byte[] QUAL1 = Bytes.toBytes("qual1");
private final static byte[] QUAL2 = Bytes.toBytes("qual2");

public static void main(String[] args) throws IOException {
Configuration conf = HBaseConfiguration.create();
HTable table = new HTable(conf, "testtable");
table.setAutoFlushTo(false);
Put put_row1 = new Put(ROW1);
put_row1.add(COLFAM1, QUAL1, Bytes.toBytes("ROW1_QUAL1_VAL"));
put_row1.add(COLFAM1, QUAL2, Bytes.toBytes("ROW1_QUAL2_VAL"));

Put put_row2 = new Put(ROW2);
put_row2.add(COLFAM1, QUAL1, Bytes.toBytes("ROW2_QUAL1_VAL"));
put_row2.add(COLFAM1, QUAL2, Bytes.toBytes("ROW2_QUAL2_VAL"));

try{
table.put(put_row1);
table.put(put_row2);
}finally{
table.close();
}
}

}

以下是过滤器测试代码,可以通过修改代码,更换过滤器来看到具体的效果:

[java] view plaincopy在CODE上查看代码片派生到我的代码片
package com.reyun.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTable;
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.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.ColumnCountGetFilter;
import org.apache.hadoop.hbase.filter.ColumnPrefixFilter;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.FirstKeyOnlyFilter;
import org.apache.hadoop.hbase.filter.InclusiveStopFilter;
import org.apache.hadoop.hbase.filter.KeyOnlyFilter;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.filter.RandomRowFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SkipFilter;
import org.apache.hadoop.hbase.filter.ValueFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.filter.WhileMatchFilter;
import org.apache.hadoop.hbase.util.Bytes;

public class HBaseScannerTest {

public static void main(String[] args) throws IOException, IllegalAccessException {
Configuration conf = HBaseConfiguration.create();
HTable table = new HTable(conf, "testtable");
table.setAutoFlushTo(false);

Scan scan1 = new Scan();
SingleColumnValueFilter scvf = new SingleColumnValueFilter(
Bytes.toBytes("colfam1"),
Bytes.toBytes("qual2"),
CompareFilter.CompareOp.NOT_EQUAL,
new SubstringComparator("BOGUS"));
scvf.setFilterIfMissing(false);
scvf.setLatestVersionOnly(true); // OK
Filter ccf = new ColumnCountGetFilter(2); // OK 如果突然发现一行中的列数超过设定的最大值时,整个扫描操作会停止
Filter vf = new ValueFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator("ROW2_QUAL1")); // OK 筛选某个(值的条件满足的)特定的单元格
Filter cpf = new ColumnPrefixFilter(Bytes.toBytes("qual2")); // OK 筛选出前缀匹配的列
Filter fkof = new FirstKeyOnlyFilter(); // OK 筛选出第一个每个第一个单元格
Filter isf = new InclusiveStopFilter(Bytes.toBytes("row1")); // OK 包含了扫描的上限在结果之内
Filter rrf = new RandomRowFilter((float) 0.8); // OK 随机选出一部分的行
Filter kof = new KeyOnlyFilter(); // OK 返回所有的行,但值全是空
Filter pf = new PrefixFilter(Bytes.toBytes("row")); // OK 筛选匹配行键的前缀成功的行
Filter rf = new RowFilter(CompareFilter.CompareOp.NOT_EQUAL, new BinaryComparator(Bytes.toBytes("row1"))); // OK 筛选出匹配的所有的行
Filter wmf = new WhileMatchFilter(rf); // OK 类似于Python itertools中的takewhile
Filter skf = new SkipFilter(vf); // OK 发现某一行中的一列需要过滤时,整个行就会被过滤掉

List<Filter> filters = new ArrayList<Filter>();
filters.add(rf);
filters.add(vf);
FilterList fl = new FilterList(FilterList.Operator.MUST_PASS_ALL, filters); // OK 综合使用多个过滤器, AND 和 OR 两种关系

scan1.
setStartRow(Bytes.toBytes("row1")).
setStopRow(Bytes.toBytes("row3")).
setFilter(scvf);
ResultScanner scanner1 = table.getScanner(scan1);

for(Result res : scanner1){
for(Cell cell : res.rawCells()){
System.out.println("KV: " + cell + ", Value: " + Bytes.toString(CellUtil.cloneValue(cell)));
}
System.out.println("------------------------------------------------------------");
}

scanner1.close();
table.close();
}



查看完整回答
反对 回复 2019-10-26
?
紫衣仙女

hbase的region是按行划分,而非按列,如果你读取指定一行的所有列数据,regionServer虽然无法保证你的所有数据都在一个HFile中,但是至少是在一个Region中。但是具体的HFile所在的hdfs的节点那就不是HBase关心的事了

查看完整回答
反对 回复 2019-10-26
?
素胚勾勒不出你

hbase的region是按行划分,而非按列,如果你读取指定一行的所有列数据,regionServer虽然无法保证你的所有数据都在一个HFile中,但是至少是在一个Region中。但是具体的HFile所在的hdfs的节点那就不是HBase关心的事了,因为HBase的存储是依赖与hdfs,所以底层存储读取的事会由NameNode操心,NameNode会考虑就近原则,而提供最高效的数据读取策略。
你的数据传输是必然,但是HBase不会计算,计算是发生在你将想要的数据获取到之后再自行进行计算的。你读取大量数据必然会有大量数据传输,HBase只是将提供了一种高效的数据读取策略,尽量减小数据传输量

查看完整回答
反对 回复 2019-10-26
?
慕斯王

hbase提供了ListFilter过滤
List<Filter> filters = new ArraList<Filter>();
Filter f1=new RowFilter(.........);
Filter f2=new AualifierFilter(................);
filters.add(f1);
filters.add(f2);
FilterList filterList=new FilterList(filters);
Scan scan=new Scan();
scan.setFilter(filterList);
ResultScanner scanner=table.getScanner(scan);
之后遍历就ok了
虽然作者很早就提问了,我还是想让更多人看到



查看完整回答
反对 回复 2019-10-26

添加回答

回复

举报

0/150
提交
取消
意见反馈 帮助中心 APP下载
官方微信