MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

HBase过滤器的综合应用与总结

2022-08-013.4k 阅读

HBase过滤器概述

HBase是一个分布式、面向列的开源数据库，运行在Hadoop分布式文件系统（HDFS）之上。在HBase中，数据以表的形式存储，表由行和列族组成，每行数据通过行键（Row Key）唯一标识。随着数据量的不断增长，在海量数据中高效地检索和获取所需数据变得至关重要，HBase过滤器应运而生。

HBase过滤器提供了一种强大且灵活的机制，用于在查询数据时对结果集进行过滤。过滤器可以在客户端发起查询时，对服务器端返回的数据进行筛选，减少网络传输的数据量，提高查询效率。过滤器基于行键、列族、列限定符、时间戳、单元格值等多种维度进行数据过滤，满足不同场景下的查询需求。

过滤器分类

按过滤对象分类

  1. 行过滤器（Row Filter）：基于行键进行过滤，用于筛选出符合特定行键模式的行数据。例如，可以根据行键前缀、行键范围等条件过滤行数据。
  2. 列过滤器（Column Filter）：针对列族和列限定符进行过滤，决定哪些列族或列限定符的数据会被返回。
  3. 单元格过滤器（Cell Filter）：作用于单元格级别，根据单元格的值、时间戳等属性进行过滤。

按操作类型分类

  1. 比较过滤器（Compare Filter）：通过比较运算来判断是否满足过滤条件，如等于、不等于、大于、小于等。
  2. 专用过滤器（Special Filter）：针对特定的过滤需求设计的过滤器，如前缀过滤器、分页过滤器等。

常用过滤器详解

比较过滤器

  1. SingleColumnValueFilter
    • 功能：根据某一列的值来过滤行。这个过滤器会检查指定列族和列限定符下的单元格值，如果值满足设定的比较条件，则该行数据会被包含在结果集中。
    • 代码示例（Java）：
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.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
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.client.Table;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;

public class SingleColumnValueFilterExample {
    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        Connection connection = ConnectionFactory.createConnection(conf);
        Table table = connection.getTable(TableName.valueOf("your_table_name"));

        // 创建SingleColumnValueFilter
        SingleColumnValueFilter filter = new SingleColumnValueFilter(
                Bytes.toBytes("cf"),
                Bytes.toBytes("col"),
                CompareFilter.CompareOp.EQUAL,
                new SubstringComparator("value_to_match")
        );

        Scan scan = new Scan();
        scan.setFilter(filter);

        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            for (Cell cell : result.rawCells()) {
                System.out.println("Row: " + Bytes.toString(CellUtil.cloneRow(cell)) +
                        ", Family: " + Bytes.toString(CellUtil.cloneFamily(cell)) +
                        ", Qualifier: " + Bytes.toString(CellUtil.cloneQualifier(cell)) +
                        ", Value: " + Bytes.toString(CellUtil.cloneValue(cell)));
            }
        }

        scanner.close();
        table.close();
        connection.close();
    }
}
  1. SingleColumnValueExcludeFilter
    • 功能：与SingleColumnValueFilter类似，但逻辑相反。它会排除掉满足指定列值条件的行，只返回不满足条件的行数据。
    • 代码示例（Java）：
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.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
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.client.Table;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueExcludeFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;

public class SingleColumnValueExcludeFilterExample {
    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        Connection connection = ConnectionFactory.createConnection(conf);
        Table table = connection.getTable(TableName.valueOf("your_table_name"));

        // 创建SingleColumnValueExcludeFilter
        SingleColumnValueExcludeFilter filter = new SingleColumnValueExcludeFilter(
                Bytes.toBytes("cf"),
                Bytes.toBytes("col"),
                CompareFilter.CompareOp.EQUAL,
                new SubstringComparator("value_to_exclude")
        );

        Scan scan = new Scan();
        scan.setFilter(filter);

        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            for (Cell cell : result.rawCells()) {
                System.out.println("Row: " + Bytes.toString(CellUtil.cloneRow(cell)) +
                        ", Family: " + Bytes.toString(CellUtil.cloneFamily(cell)) +
                        ", Qualifier: " + Bytes.toString(CellUtil.cloneQualifier(cell)) +
                        ", Value: " + Bytes.toString(CellUtil.cloneValue(cell)));
            }
        }

        scanner.close();
        table.close();
        connection.close();
    }
}
  1. RowFilter
    • 功能：基于行键进行过滤。通过设定比较器和比较操作，判断行键是否满足条件，从而决定该行是否被包含在结果集中。
    • 代码示例（Java）：
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.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
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.client.Table;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;

public class RowFilterExample {
    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        Connection connection = ConnectionFactory.createConnection(conf);
        Table table = connection.getTable(TableName.valueOf("your_table_name"));

        // 创建RowFilter
        RowFilter filter = new RowFilter(
                CompareFilter.CompareOp.EQUAL,
                new SubstringComparator("row_key_prefix")
        );

        Scan scan = new Scan();
        scan.setFilter(filter);

        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            for (Cell cell : result.rawCells()) {
                System.out.println("Row: " + Bytes.toString(CellUtil.cloneRow(cell)) +
                        ", Family: " + Bytes.toString(CellUtil.cloneFamily(cell)) +
                        ", Qualifier: " + Bytes.toString(CellUtil.cloneQualifier(cell)) +
                        ", Value: " + Bytes.toString(CellUtil.cloneValue(cell)));
            }
        }

        scanner.close();
        table.close();
        connection.close();
    }
}
  1. ColumnPrefixFilter
    • 功能：根据列限定符的前缀进行过滤。它会返回列限定符以指定前缀开头的列数据。
    • 代码示例（Java）：
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.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
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.client.Table;
import org.apache.hadoop.hbase.filter.ColumnPrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;

public class ColumnPrefixFilterExample {
    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        Connection connection = ConnectionFactory.createConnection(conf);
        Table table = connection.getTable(TableName.valueOf("your_table_name"));

        // 创建ColumnPrefixFilter
        ColumnPrefixFilter filter = new ColumnPrefixFilter(Bytes.toBytes("col_prefix"));

        Scan scan = new Scan();
        scan.setFilter(filter);

        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            for (Cell cell : result.rawCells()) {
                System.out.println("Row: " + Bytes.toString(CellUtil.cloneRow(cell)) +
                        ", Family: " + Bytes.toString(CellUtil.cloneFamily(cell)) +
                        ", Qualifier: " + Bytes.toString(CellUtil.cloneQualifier(cell)) +
                        ", Value: " + Bytes.toString(CellUtil.cloneValue(cell)));
            }
        }

        scanner.close();
        table.close();
        connection.close();
    }
}

专用过滤器

  1. PrefixFilter
    • 功能：这是一种行过滤器，用于匹配行键以指定前缀开头的行。在需要按行键前缀快速定位大量相关行数据时非常有用。
    • 代码示例（Java）：
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.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
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.client.Table;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;

public class PrefixFilterExample {
    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        Connection connection = ConnectionFactory.createConnection(conf);
        Table table = connection.getTable(TableName.valueOf("your_table_name"));

        // 创建PrefixFilter
        PrefixFilter filter = new PrefixFilter(Bytes.toBytes("row_prefix"));

        Scan scan = new Scan();
        scan.setFilter(filter);

        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            for (Cell cell : result.rawCells()) {
                System.out.println("Row: " + Bytes.toString(CellUtil.cloneRow(cell)) +
                        ", Family: " + Bytes.toString(CellUtil.cloneFamily(cell)) +
                        ", Qualifier: " + Bytes.toString(CellUtil.cloneQualifier(cell)) +
                        ", Value: " + Bytes.toString(CellUtil.cloneValue(cell)));
            }
        }

        scanner.close();
        table.close();
        connection.close();
    }
}
  1. PageFilter
    • 功能：用于实现分页查询。通过设定每页返回的行数，控制查询结果集的大小，方便在海量数据中逐页获取数据。
    • 代码示例（Java）：
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.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
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.client.Table;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.util.Bytes;

public class PageFilterExample {
    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        Connection connection = ConnectionFactory.createConnection(conf);
        Table table = connection.getTable(TableName.valueOf("your_table_name"));

        // 创建PageFilter，每页返回10行
        PageFilter filter = new PageFilter(10);

        Scan scan = new Scan();
        scan.setFilter(filter);

        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            for (Cell cell : result.rawCells()) {
                System.out.println("Row: " + Bytes.toString(CellUtil.cloneRow(cell)) +
                        ", Family: " + Bytes.toString(CellUtil.cloneFamily(cell)) +
                        ", Qualifier: " + Bytes.toString(CellUtil.cloneQualifier(cell)) +
                        ", Value: " + Bytes.toString(CellUtil.cloneValue(cell)));
            }
        }

        scanner.close();
        table.close();
        connection.close();
    }
}

过滤器的组合使用

在实际应用中，往往需要多个过滤器协同工作来满足复杂的查询需求。HBase支持将多个过滤器组合使用，通过FilterList类来实现。FilterList可以包含多个过滤器，并且可以设定过滤器之间的关系，如AND关系或OR关系。

AND关系的过滤器组合

当使用AND关系组合过滤器时，只有当所有过滤器的条件都满足时，该行数据才会被包含在结果集中。

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.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
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.client.Table;
import org.apache.hadoop.hbase.filter.ColumnPrefixFilter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;

public class AndFilterListExample {
    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        Connection connection = ConnectionFactory.createConnection(conf);
        Table table = connection.getTable(TableName.valueOf("your_table_name"));

        // 创建PrefixFilter
        PrefixFilter rowPrefixFilter = new PrefixFilter(Bytes.toBytes("row_prefix"));
        // 创建ColumnPrefixFilter
        ColumnPrefixFilter colPrefixFilter = new ColumnPrefixFilter(Bytes.toBytes("col_prefix"));

        // 创建FilterList，设定为AND关系
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
        filterList.addFilter(rowPrefixFilter);
        filterList.addFilter(colPrefixFilter);

        Scan scan = new Scan();
        scan.setFilter(filterList);

        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            for (Cell cell : result.rawCells()) {
                System.out.println("Row: " + Bytes.toString(CellUtil.cloneRow(cell)) +
                        ", Family: " + Bytes.toString(CellUtil.cloneFamily(cell)) +
                        ", Qualifier: " + Bytes.toString(CellUtil.cloneQualifier(cell)) +
                        ", Value: " + Bytes.toString(CellUtil.cloneValue(cell)));
            }
        }

        scanner.close();
        table.close();
        connection.close();
    }
}

OR关系的过滤器组合

使用OR关系组合过滤器时，只要有一个过滤器的条件满足，该行数据就会被包含在结果集中。

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.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
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.client.Table;
import org.apache.hadoop.hbase.filter.ColumnPrefixFilter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.util.Bytes;

public class OrFilterListExample {
    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        Connection connection = ConnectionFactory.createConnection(conf);
        Table table = connection.getTable(TableName.valueOf("your_table_name"));

        // 创建PrefixFilter
        PrefixFilter rowPrefixFilter = new PrefixFilter(Bytes.toBytes("row_prefix"));
        // 创建ColumnPrefixFilter
        ColumnPrefixFilter colPrefixFilter = new ColumnPrefixFilter(Bytes.toBytes("col_prefix"));

        // 创建FilterList，设定为OR关系
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ONE);
        filterList.addFilter(rowPrefixFilter);
        filterList.addFilter(colPrefixFilter);

        Scan scan = new Scan();
        scan.setFilter(filterList);

        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            for (Cell cell : result.rawCells()) {
                System.out.println("Row: " + Bytes.toString(CellUtil.cloneRow(cell)) +
                        ", Family: " + Bytes.toString(CellUtil.cloneFamily(cell)) +
                        ", Qualifier: " + Bytes.toString(CellUtil.cloneQualifier(cell)) +
                        ", Value: " + Bytes.toString(CellUtil.cloneValue(cell)));
            }
        }

        scanner.close();
        table.close();
        connection.close();
    }
}

过滤器性能优化

减少过滤器数量

尽量避免使用过多的过滤器，因为每个过滤器都会增加处理开销。在设计过滤器组合时，仔细评估是否真的需要每个过滤器，尝试合并或简化过滤器逻辑，以减少整体的过滤操作。

合理选择过滤器

根据查询需求，选择最合适的过滤器。例如，如果只是按行键前缀查询，使用PrefixFilter会比使用复杂的RowFilter更高效。对于列相关的过滤，优先使用专门的列过滤器，而不是在单元格过滤器中通过复杂逻辑判断列。

利用缓存

HBase支持查询结果缓存，合理利用缓存可以减少重复查询相同数据的开销。在客户端设置合适的缓存参数，使得经常查询的数据可以快速从缓存中获取，而不需要每次都通过过滤器进行实时筛选。

预分区和数据布局优化

在表设计阶段，通过合理的预分区和数据布局，可以使过滤器在查询时能够更快地定位到所需数据。例如，根据行键的分布规律进行预分区，让相关的数据存储在同一区域，减少过滤器在多个区域间的扫描。

应用场景

日志数据分析

在日志数据存储于HBase的场景中，日志数据通常按时间戳作为行键。可以使用RowFilter结合时间范围比较器，快速筛选出特定时间段内的日志数据。同时，通过SingleColumnValueFilter可以根据日志级别等列值过滤出关键日志信息。

用户画像查询

假设用户画像数据存储在HBase中，行键为用户ID。可以通过PrefixFilter按用户ID前缀快速定位某一类用户（如某个地区的用户），再结合ColumnPrefixFilter获取特定属性列族下的相关属性信息，以构建和查询用户画像。

实时监控数据处理

在实时监控系统中，监控数据不断写入HBase。使用PageFilter可以实现分页展示监控数据，方便运维人员逐页查看最新的监控信息。同时，利用SingleColumnValueExcludeFilter可以排除掉一些正常状态的数据，只关注异常状态的监控指标。