做HBase和MapReduce联调的时候，我见过太多这样的场景：一个统计任务，逻辑明明很简单，就是按条件筛一批数据再汇总，结果跑了一个多小时还没结束，打开监控一看，Mapper读了几千万行，实际上真正用到的只有几万行。问题基本都出在同一个地方——数据读取阶段没有做任何收敛，把整个表当成了一个巨大的输入管道，让所有数据都涌进了计算框架。
这篇东西想聊聊一个非常具体的组合：HBase过滤器怎么用，以及MapReduce任务里数据输出量到底怎么控。不是泛泛讲API，而是把我在实际项目里反复调过的选型、参数、踩坑经验整理出来，给正要写类似任务的读者提供一条能直接落地的参照路径。
1. 为什么把过滤器和MapReduce数据输出量放在一起拆
1.1 HBase和MapReduce之间的天然矛盾
HBase本质上是一个分布式列式KV存储，数据按照行键有序分布在各Region上，读取的最小单元是一行里的某个单元格。MapReduce则期望输入是“可切分的记录流”，每个Mapper处理一批(key, value)对，然后洗牌、归并、输出。两者结合时最隐蔽的问题是：HBase向MapReduce暴露的数据粒度是“整行结果”，而不是“你真正关心的那几列”。
举个例子，一张订单表一行有20个字段，你只关心“订单状态”和“金额”两列，但默认情况下一个Scan会把整行的所有列全部返回。列不需要的全部经过网络传输、经过反序列化、进入Mapper堆内存，最终在Mapper里被丢弃。这种浪费不会让任务直接报错，但会表现为IO上涨、GC频繁、运行时间线性拉长。
1.2 “先全部读出来再过滤”是默认陷阱
很多同学习惯在Mapper的map方法里写这种逻辑：
if (!"PAID".equals(Bytes.toString(row.getValue(CF, COL)))) { return; }这当然能过滤数据，但过滤发生得太晚了。数据已经从RegionServer传输到了客户端，遍历了ResultScanner的所有结果，业务代码才决定丢弃。更糟的是，HBase的Scan默认会尽可能把RegionServer上的数据搬回来，搬多少取决于Caching参数，和你的过滤条件没有任何关系。
我称之为“迟到过滤”，它和“前置过滤”的差别，在数据量小的时候看不出来，一旦单表行数过亿、单行列数超过十几个，任务时间差距会非常明显。
1.3 观察输出量控制效果的三项指标
衡量一个MR任务的数据读取是否合理，我建议盯三个数字：
- RegionServer被扫描的行数：即HBase侧实际返回给客户端的行数，理想情况下等于过滤后的目标行数，而不是全表行数。
- Map输入记录的字节数与行数：看的是InputFormat真正构造了多少次map调用。
- Shuffle阶段的数据量：Map输出经过分区、排序后传输给Reducer的数据量，这个数字最能反映中间数据是否膨胀。
如果这三个数字里任何一个明显偏离“业务最终结果”一个量级以上，那基本可以断定写的是“全表扫描型任务”。
2. HBase过滤器机制核心梳理
2.1 过滤器在整套链路中的位置
HBase的过滤器在服务端执行。RegionServer扫描本地StoreFile时，会逐行检查过滤器条件，只有通过的记录才会被写回客户端。这意味着过滤器越早介入，数据传输量越小。它和数据库里的“下推”思路一样——把计算尽量放到数据所在的位置，而不是把数据搬到别处再算。
所以在HBase加MapReduce的场景里，Scan阶段必须做三件套：
- 用startRow/endRow限制行键范围，这是最廉价的裁剪；
- 用addFamily/addColumn裁剪列，减少单行返回的数据宽度；
- 用Filter裁剪行，减少返回给客户端的行数。
三者组合使用，才是完整的前置过滤。
2.2 常用过滤器分类与选型
HBase自带的常用过滤器有以下几类，选型时可以按用途对号入座。
| 过滤器 | 用途 | 典型场景 |
|---|---|---|
| RowFilter | 对行键做比较或正则过滤 | 行键包含日期/前缀/区间 |
| PrefixFilter | 行键前缀匹配 | 按业务维度聚合的行键设计 |
| ColumnPrefixFilter | 按列限定前缀返回 | 列族内多列但只取特定系列 |
| SingleColumnValueFilter | 根据某列的值决定整行是否返回 | 状态字段过滤，如订单状态 |
| ValueFilter | 对某个单元格值做过滤 | 全列值范围匹配 |
| QualifierFilter | 对列名过滤 | 只看指定列 |
| KeyOnlyFilter | 只返回行键和列名，不返回值 | 快速统计行数 |
| FirstKeyOnlyFilter | 只返回每行第一个单元格 | 快速判断行是否存在 |
| TimestampsFilter | 只返回指定时间戳版本 | 历史版本清理 |
实际用得最多的是SingleColumnValueFilter。注意它有几个很容易坑到人的行为：默认只检查第一个匹配到的单元格。如果要让整行返回必须做成“列与条件的配对检查”，推荐把它和DependentColumnFilter配合，或者用FilterList组合。
2.3 FilterList组合与执行顺序
多个过滤条件不要各自setFilter，HBase的Scan一次只能设置一个Filter。正确做法是把所有过滤器放进FilterList。
FilterList list = new FilterList(FilterList.Operator.MUST_PASS_ALL); list.addFilter(new SingleColumnValueFilter( CF, Bytes.toBytes("status"), CompareOperator.EQUAL, Bytes.toBytes("PAID"))); list.addFilter(new ColumnPrefixFilter(Bytes.toBytes("amount"))); Scan scan = new Scan(); scan.setFilter(list);两个操作符语义要注意：
- MUST_PASS_ALL：所有条件都满足才返回，相当于AND；
- MUST_PASS_ONE：任意一个条件满足就返回，相当于OR。
执行顺序上，FilterList会按照添加顺序逐个判断，所以要把区分度高、计算量小的条件放在前面。比如先用RowFilter把行键区间之外的扫掉，再用SingleColumnValueFilter做精确值过滤，比反过来能省不少无谓的单元格读取。
2.4 Caching、Batch与SmallScan的参数细节
Scan从RegionServer拉数据是按“批次”拉的。这两个参数直接影响任务的内存和RPC次数：
- setCaching(n)：每次RPC向服务端请求的行数，默认100。MR场景建议调到500-1000，但要结合单行大小，单行1KB和单行100KB的Memory压力完全不同。
- setBatch(n)：每次RPC请求返回的列单元格数量。如果一个表行宽很大，几十列、上百列，Caching又很大，一次返回的Result对象可能把客户端堆撑爆。此时要设置Batch，比如20或50，但Batch会切分一行数据返回，破坏整行的完整性。
MapReduce下更推荐用现成的TableMapReduceUtil起任务，内部已经处理了Caching和Batch的初始化，再用SmallScan参数控制每个Mapper的扫描范围：
TableMapReduceUtil.initTableMapperJob( tableName, scan, MapperClass.class, ImmutableBytesWritable.class, Result.class, job, false, null, 100);最后那个100是分片大小，单位是“每个Mapper处理的Region数量”的约束。设得越小，单Mapper扫描的数据量越少，适合表大、Region多、过滤条件强的任务。
3. MapReduce侧数据输出量控制实操
3.1 先按行键区间收敛，再谈条件过滤
HBase最擅长的查询方式是按行键定位。如果你有明确的业务键——比如用户ID、日期前缀——一定要把它编码进行键，并利用startRow和endRow缩小Scan范围。这个步骤能省掉的服务端扫描量，往往比过滤器还多。
一个典型的行键设计是：用户ID_日期。在Scan里可以这样收敛：
byte[] startRow = Bytes.toBytes("user_" + userId + "_20240101"); byte[] endRow = Bytes.toBytes("user_" + userId + "_20240102"); scan.withStartRow(startRow); scan.withStopRow(endRow);endRow是不包含的，所以要取“下一天的开始”作为结束边界。这一招直接把要扫描的Region限定在几个甚至一个Range内，RegionServer连打开StoreFile的成本都省了一半。
3.2 把过滤逻辑下推给RegionServer
Mapper里的Java代码不要做条件判断，条件判断全部尝试翻译成Filter。最常见的几个翻译套路：
- 等值条件 → SingleColumnValueFilter；
- 范围条件 → RowFilter或scan的行键范围；
- 多条件的组合 → FilterList；
- “只需要统计行数，不需要明细” → KeyOnlyFilter + 自定义空Evaluator。
有人会担心自定义Filter太复杂，其实HBase支持继承FilterBase来写轻量级过滤逻辑。比如要过滤“金额字段必须大于1000且状态是PAID”，纯用SingleColumnValueFilter也能表达，但需要两列联动时，直接在自定义Filter里重写filterRowCells更直观：
public class MyOrderFilter extends FilterBase { @Override public ReturnCode filterCell(Cell c) { // 可以拿到每个单元格的列族、列名、值 return ReturnCode.INCLUDE; } @Override public boolean filterRow() { // 行级过滤在最后统一裁决 return false; } }核心原则是：真正发生过滤的位置应该靠近StoreFile，越靠近存储层，节省的网络IO和内存越多。
3.3 列裁剪与版本裁剪
添加列裁剪的三个途径，按优先级排：
- 只addColumn需要的列，其他列根本不出StoreFile；
- 设置setMaxVersions(1)，只读最新版本，避免HBase默认把所有版本都返回；
- 用setTimeRange限定时间戳窗口，这不仅是过滤，还能让RegionServer跳过不必要的版本合并。
scan.addColumn(CF, Bytes.toBytes("status")); scan.addColumn(CF, Bytes.toBytes("amount")); scan.setMaxVersions(1); scan.setTimeRange(startTs, endTs);这里的坑在于，setMaxVersions并不限制HBase扫描引擎读多少个版本，它只影响返回给客户端的版本数。真正想减少底层读放大，setTimeRange更有效。两个配合用效果最好。
3.4 用计数器先估算输出规模
控制输出量的第一步是知道“到底会有多少输出”。我一般会在Mapper里加一个自定义Counter，统计三类数量：读取行数、通过行数、输出字节数。
context.getCounter("MY_PROFILE", "SCAN_ROWS").increment(1);跑了10%数据后看一眼任务页面的Counter数值，就能推算出全量数据量级。如果发现读取行数远大于通过行数，说明过滤条件没下推到位，赶紧停任务改Scan，而不是等它跑完再优化。
MapReduce任务不像在线接口，跑挂了可以随时重来，但一次全表扫描的代价可能是几小时的集群资源，所以“先评估后执行”永远值得。
3.5 Reducer侧进一步收敛输出量
Map阶段读到的数据量已经收敛后，Reducer侧还可能出现二次膨胀。常见的情况是：每条记录在map输出时key粒度太细，导致大量细碎记录涌入同一个Reducer，排序、合并、落盘的文件体积成倍增加。
对策有几个：
- key粒度提粗，比如把订单ID换成“用户ID_日期”，让一个重要维度聚合；
- 在Mapper端做Combiner，提前聚合同key数据；
- 输出文件序列化和压缩。写HDFS时，建议用SequenceFile或Parquet，并开Snappy压缩，减少落盘体积。
job.setOutputFormatClass(SequenceFileOutputFormat.class); FileOutputFormat.setOutputCompressorClass(job, SnappyCodec.class);这一套组合下来，最终写出的数据量，常常能比最原始的版本缩小几十倍。
4. 实战：一个模拟订单分析任务的改造前后
4.1 表结构与任务需求
给一个模拟订单系统X的案例。订单表结构如下：
- 表名：orders_log
- 行键：
customerNo_timestamp，例如C1024_1700000000000 - 列族：info
- 列：order_id、customer_no、status、amount、create_time
需求是：统计2024年1月内，状态为“PAID”的订单总金额，按客户维度输出客户号与金额汇总。简单来说就是一个带过滤条件的分组聚合。
这个任务如果直接写全表扫描，会读orders_log里所有行、所有列、所有版本，然后通过到客户端后再用Java条件过滤。数据规模假设是10亿行，其中符合条件的大约1亿行。
4.2 第一版未优化任务的实现
最初的实现大概是这样的：
public class BadOrderMapper extends TableMapper<Text, LongWritable> { @Override protected void map(ImmutableBytesWritable key, Result value, Context context) { String status = Bytes.toString(value.getValue(CF, Bytes.toBytes("status"))); if ("PAID".equals(status)) { byte[] amount = value.getValue(CF, Bytes.toBytes("amount")); String customer = Bytes.toString(key.get()); context.write(new Text(customer.split("_")[0]), new LongWritable(Bytes.toLong(amount))); } } }Scan没有设置任何过滤器，Caching还是默认值。任务的实际表现是：Mapper跑了接近3小时，输入记录数等于全表行数10亿。每一条map调用都会创建一个Result对象，即使大部分都被Java代码丢弃。GC时间占了任务时长的接近三分之一，集群IO长时间处于高位。
4.3 第二版过滤下推的实现
改造后的Scan：
Scan scan = new Scan(); FilterList filters = new FilterList(FilterList.Operator.MUST_PASS_ALL); filters.addFilter(new SingleColumnValueFilter( CF, Bytes.toBytes("status"), CompareOperator.EQUAL, Bytes.toBytes("PAID"))); filters.addFilter(new ColumnPrefixFilter(Bytes.toBytes("amount"))); filters.addFilter(new ColumnPrefixFilter(Bytes.toBytes("customer_no"))); scan.setFilter(filters); scan.setMaxVersions(1); scan.addColumn(CF, Bytes.toBytes("status")); scan.addColumn(CF, Bytes.toBytes("amount")); scan.addColumn(CF, Bytes.toBytes("customer_no")); scan.setCaching(500); scan.setBatch(100);这里把过滤逻辑下沉到了RegionServer，同时通过addColumn只取三列，并且用ColumnPrefixFilter只保留需要的列前缀。Mapper里的Java判断其实已经冗余，但保留也无妨，作为最后一道保险。
第二版任务跑完，读取行数变成了大约1亿行，Map输入字节数下降了80%以上，任务时长从3小时缩短到40分钟。Reducer侧输出量也从原来的几十GB，降到了不足2GB。
4.4 执行数据对比与效果分析
| 指标 | 未优化版本 | 过滤下推版本 |
|---|---|---|
| RegionServer返回行数 | 10亿 | 约1亿 |
| 单行返回列数 | 5列全量 | 3列 |
| Map数据源字节数 | 约2.5TB | 约400GB |
| 任务总耗时 | 约3小时 | 约42分钟 |
| GC时间占比 | 约30% | 约8% |
| 最终输出文件大小 | 约35GB | 约1.8GB |
数据差异大的原因很简单：未优化版本做了大量无用的数据传输、反序列化和对象创建；优化版本让RegionServer先把数据筛掉，Mapper只处理业务真正关心的记录。
5. 常见问题与排查实录
5.1 过滤器写了，但任务还是全表扫描
这是最高频的问题。排查步骤：
- 第一步，确认Filter真的set进Scan了，而不是只add到FilterList但没setFilter。这个错误我见过不止一次。
- 第二步，确认FilterList里的SingleColumnValueFilter指定了正确的列族和列名。列族传错会导致过滤条件永远不匹配，HBase可能返回空，也可能返回全量，取决于Filter的处理逻辑。
- 第三步，确认过滤条件是否需要设为FILTER_BY_KEY。SingleColumnValueFilter默认有一个setFilterIfMissing属性，如果某行没有这个列，默认该行会被排除。如果你希望“缺列也返回”，需要显式设false。反向的坑是，把它设置成true后，过滤效果看起来正常，但数据量不符合预期，排查时容易误判。
建议排查时用一个小的SampleScan先在HBase Shell里跑一遍Filter，确认返回行数符合预期，再塞进MR任务。
5.2 Mapper内存溢出
加了过滤器后，RegionServer返回的数据少了很多，但单行数据如果特别宽，仍然可能出现堆溢出。原因是Result对象一次性加载一行的所有单元格，如果Caching设置得太大，一个Result里可能有几十上百个Cell。
解决办法：
- 调小Caching，比如从1000降到200；
- 调小Batch，强制一行被拆成多个Result；
- 增加Mapper的Xmx，比如-Dmapreduce.map.memory.mb=2048。
重点还是优化Scan的列裁剪，让单行数据本身变窄，而不是靠内存硬扛。
5.3 输出数据量不降反增
有人改了过滤器后，发现输出文件变多了。常见原因有两个：
- Mapper的输出的key粒度太细，导致Reducer数量被强行推高，每个Reducer只处理少量数据，但各自写一个文件，文件数量增加，总字节数反而可能变大；
- SequenceFile/MapFile格式未设置压缩。
处理办法是手工设置Reducer数量：job.setNumReduceTasks(具体数值)，根据最终输出量预估，比如1GB数据建议10-20个Reducer，同时开启Snappy压缩，保证文件体积和数量都可控。
5.4 数据倾斜导致个别Reducer过慢
过滤后数据量小了，但分布不均。比如按客户聚合，大客户订单多，单个Reducer承担了绝大多数数据。
优化办法是在Mapper端先做局部聚合（Combiner），把同一客户的数据先合并成一条记录，再将客户号作为key输出。Reducer侧只做最终汇总，负载会平滑很多。
注意Combiner的combine逻辑必须满足“可重复应用”的性质，否则会改变最终结果。
6. 调优参数速查与个人心得
6.1 参数速查表
| 场景 | 推荐配置 |
|---|---|
| 常规MR读HBase | TableMapReduceUtil.initTableMapperJob + Scan.setCaching(500) |
| 行宽较大（单行>50列） | 增加Batch、减少Caching，列裁剪优先 |
| 行数极多（>1亿） | 加FilterList组合，startRow/endRow收敛，分片SmallScan=50 |
| 只统计行数 | KeyOnlyFilter + FirstKeyOnlyFilter组合 |
| 输出到HDFS | SequenceFile + Snappy |
| 避免全表扫描 | 检查Scanner返回行数与全表比例 |
| 中间数据过大 | Mapper端Combiner，key粒度适度调粗 |
6.2 三条亲手踩过的经验
第一，过滤器不是越多越好。FilterList里塞十个条件，每个条件都会在RegionServer逐行调用，条件多了反而比不带过滤器更慢。筛选条件的优先级应该是：行键范围 > 列裁剪 > 行过滤器 > 值过滤器，能在Scan层面解决的，绝不要靠Filter。
第二，任务跑之前一定要看“预计扫描行数”。HBase的RegionServer日志和监控面板上有相关指标，MR任务启动前，可以先通过HBase Shell跑一下同样的Scan，用scan(...)返回的ResultScanner迭代一次，打印行数，判断数据量级。这一步只要几秒钟，却能在任务启动前挡住绝大多数低效方案。
第三，数据输出量不光看行数，还看字节数。很多时候过滤后的行数很少，但每行的列值都很大（比如画像文本、日志详情），实际字节数依然爆炸。遇到这种情况，要在列裁剪上下功夫，慎重把大字段留在Result里，必要时使用值截断过滤器或改成HDFS上的明细表关联读取。
做HBase和MR任务这么久，我的体验是：大部分性能问题都不是框架不够快，而是业务数据在链路上多走了不该走的路。过滤器加在Scan上、列裁剪做在前面、聚合下沉到Mapper端，同一套逻辑往往能从小时级优化到分钟级。希望这篇整理能帮你少踩几个类似的坑。
