hbase 源代码解析（21） 自定义过滤器

filter 需要实现Filter 或者继承FilterBase

1. @InterfaceAudience.Public
2. @InterfaceStability.Stable
3. public abstract class Filter {
4. //返回码
5. public enum ReturnCode {
6. INCLUDE, //结果中包含着一样
7. INCLUDE_AND_NEXT_COL, //包含着这样一行，跳到下一行比较
8. SKIP, //跳到下一个keyvalue 并进行处理
9. NEXT_COL, //跳过当前一col
10. NEXT_ROW, //跳过当前一行
11. SEEK_NEXT_USING_HINT, //跳到下一个满足地方，需要调用getNextKeyHint（）
12. }
13. protected transient boolean reversed;
15. abstract public void reset() throws IOException;
17. //判断行健是否满足，不满足可以跳过，避免其他检查：比如前缀过滤器
18. abstract public boolean filterRowKey(byte[] buffer, int offset, int length) throws IOException;
19. //这个过滤器可以提前结束
20. abstract public boolean filterAllRemaining() throws IOException;
21. //对cell处理，
22. abstract public ReturnCode filterKeyValue(final Cell v) throws IOException;
23. abstract public Cell transformCell(final Cell v) throws IOException;
24. @Deprecated // use Cell transformCell(final Cell)
25. abstract public KeyValue transform(final KeyValue currentKV) throws IOException;
26. //经过前面处理后，如果还有数据，将对当前行一起处理， 比如依赖过去器
27. abstract public void filterRowCells(List<Cell> kvs) throws IOException;
28. abstract public boolean hasFilterRow()；
29. //经过这么多流程如果还有数据，会去检查一下数据的要求。比如pagefilter 是否已经够一页了
30. abstract public boolean filterRow() throws IOException;
31. @Deprecated
32. abstract public KeyValue getNextKeyHint(final KeyValue currentKV) throws IOException;
33. abstract public Cell getNextCellHint(final Cell currentKV) throws IOException;
34. abstract public boolean isFamilyEssential(byte[] name) throws IOException;
35. abstract public byte[] toByteArray() throws IOException;
36. public static Filter parseFrom(final byte [] pbBytes) throws DeserializationException {
37. throw new DeserializationException(
38. "parseFrom called on base Filter, but should be called on derived type");
39. }
40. abstract boolean areSerializedFieldsEqual(Filter other);
41. public void setReversed(boolean reversed) {
42. this.reversed = reversed;
43. }
44. public boolean isReversed() {
45. return this.reversed;
46. }
47. }

流程如下：

 

scan，或者get是调用的入口

基本流程就是下面 一样，但是没有看到fiterKeyValue，如果你们找到了，告诉我一声。

1. private boolean nextInternal(List<Cell> results, ScannerContext scannerContext)
2. throws IOException {
5. while (true) {
6.  
7. boolean stopRow = isStopRow(currentRow, offset, length);
9. boolean hasFilterRow = this.filter != null && this.filter.hasFilterRow();
12. if (hasFilterRow) {
13. if (LOG.isTraceEnabled()) {
14. LOG.trace("filter#hasFilterRow is true which prevents partial results from being "
15. + " formed. Changing scope of limits that may create partials");
16. }
17. scannerContext.setSizeLimitScope(LimitScope.BETWEEN_ROWS);
18. scannerContext.setTimeLimitScope(LimitScope.BETWEEN_ROWS);
19. }
21. if (filterRowKey(currentRow, offset, length)) {
22. incrementCountOfRowsFilteredMetric(scannerContext);
23. // early check, see HBASE-16296
24. //filterAllRemaining 实际调用
25. if (isFilterDoneInternal()) {
26. return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
27. }
28. incrementCountOfRowsScannedMetric(scannerContext);
29. //里面会调用filter.reset();
30. boolean moreRows = nextRow(scannerContext, currentRow, offset, length);
31. if (!moreRows) {
32. return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
33. }
34. results.clear();
35. continue;
36. }
38. populateResult(results, this.storeHeap, scannerContext, currentRow, offset, length);
40. Cell nextKv = this.storeHeap.peek();
41. stopRow = nextKv == null ||
42. isStopRow(nextKv.getRowArray(), nextKv.getRowOffset(), nextKv.getRowLength());
44. final boolean isEmptyRow = results.isEmpty();
45. 1. FilterWrapper.FilterRowRetCode ret = FilterWrapper.FilterRowRetCode.NOT_CALLED;
46. if (hasFilterRow) {
47. //会调用filterRowCells（List） 和filterRowCells（cell）
48. ret = filter.filterRowCellsWithRet(results);
50. long timeProgress = scannerContext.getTimeProgress();
51. if (scannerContext.getKeepProgress()) {
52. scannerContext.setProgress(initialBatchProgress, initialSizeProgress,
53. initialTimeProgress);
54. } else {
55. scannerContext.clearProgress();
56. }
57. scannerContext.setTimeProgress(timeProgress);
58. scannerContext.incrementBatchProgress(results.size());
59. for (Cell cell : results) {
60. scannerContext.incrementSizeProgress(CellUtil.estimatedHeapSizeOfWithoutTags(cell));
61. }
62. }
64. if (isEmptyRow || ret == FilterWrapper.FilterRowRetCode.EXCLUDE || filterRow()) {
65. incrementCountOfRowsFilteredMetric(scannerContext);
66. results.clear();
67. boolean moreRows = nextRow(scannerContext, currentRow, offset, length);
68. if (!moreRows) {
69. return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
70. }
72. if (!stopRow) continue;
73. return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
74. }
76. if (this.joinedHeap != null) {
77. boolean mayHaveData = joinedHeapMayHaveData(currentRow, offset, length);
78. if (mayHaveData) {
79. joinedContinuationRow = current;
80. populateFromJoinedHeap(results, scannerContext);
82. if (scannerContext.checkAnyLimitReached(LimitScope.BETWEEN_CELLS)) {
83. return true;
84. }
85. }
86. }
87. } else {
88. populateFromJoinedHeap(results, scannerContext);
89. if (scannerContext.checkAnyLimitReached(LimitScope.BETWEEN_CELLS)) {
90. return true;
91. }
92. }
95. if (stopRow) {
96. return scannerContext.setScannerState(NextState.NO_MORE_VALUES).hasMoreValues();
97. } else {
98. return scannerContext.setScannerState(NextState.MORE_VALUES).hasMoreValues();
99. }
100. }
101. }

就这样。

自定义完成后，打成jar ，需要export export HBASE_CLAPSS 

或者将jar 放到hbase 的安装目录的lib下面，重启hbase