4、增量计算加速技术：增量索引、增量聚合、增量关联、增量去重

好，咱们进入正题。增量计算加速，说白了就是「只算变化的部分」。

我刚开始做日志处理那会儿，每天全量跑一次，数据量一上来就跑不动了。后来才明白——大部分数据是没变的，你何必每次都从头算？

这一章，我带你拆解四个核心加速手段：增量索引、增量聚合、增量关联、增量去重。它们不是孤立的技术，而是一套组合拳。

4.1 增量索引：别重建，只追加

索引是查询加速的命根子。但全量重建索引，数据量一大就扛不住。

增量索引的思路很简单：新来的数据，追加到新索引段里，而不是把整个索引推倒重来。

实现增量索引，一般分三步：

1. 接收增量数据——只拿新增或修改的日志行
2. 构建小索引段——在内存里生成倒排索引，刷到磁盘
3. 段合并——后台线程把小段合并成大段，控制段数量

代码示例（伪代码，展示增量索引的核心逻辑）：

// 增量索引构建器
class IncrementalIndexBuilder {
    private List<Segment> segments = new ArrayList<>();
    
    public void addDocuments(List<Document> docs) {
        // 1. 在内存中构建小段
        Segment newSeg = buildSegment(docs);
        
        // 2. 刷到磁盘
        flushToDisk(newSeg);
        
        // 3. 加入段列表
        segments.add(newSeg);
        
        // 4. 触发异步合并（段数超过阈值时）
        if (segments.size() > MERGE_THRESHOLD) {
            triggerAsyncMerge();
        }
    }
    
    public List<Result> search(Query query) {
        // 查询时，遍历所有段，合并结果
        List<Result> results = new ArrayList<>();
        for (Segment seg : segments) {
            results.addAll(seg.search(query));
        }
        return mergeAndSort(results);
    }
}

4.2 增量聚合：窗口 + 累加器

日志处理里，聚合是家常便饭。按小时统计 PV、按 IP 统计错误数……

全量聚合的做法是：把历史数据全拉出来，重新算一遍。这太傻了。

增量聚合的核心是「状态保持」。你想想看，昨天的聚合结果已经算好了，今天只需要把新来的数据合并进去就行。

常见的增量聚合模式有两种：

模式	适用场景	实现方式	我踩过的坑
滑动窗口	实时统计（最近 5 分钟、1 小时）	Flink 的 Tumbling/Sliding Window	窗口边界处理不好会丢数据
累加器模式	长期累计（日活、累计 UV）	Redis + 原子计数器	并发写导致计数不准

举个例子，用 Flink 做增量聚合：

// Flink 增量聚合示例
DataStream<LogEvent> logStream = env.addSource(kafkaConsumer);

// 按 key 分组，开 1 小时的滑动窗口，每 5 分钟滑动一次
logStream
    .keyBy(event -> event.getIp())
    .window(SlidingProcessingTimeWindows.of(
        Time.hours(1), Time.minutes(5)))
    .aggregate(new CountAggregator())
    .addSink(new RedisSink());

// 自定义累加器
public class CountAggregator 
    implements AggregateFunction<LogEvent, Long, Long> {
    
    @Override
    public Long createAccumulator() {
        return 0L;  // 初始值
    }
    
    @Override
    public Long add(LogEvent value, Long accumulator) {
        return accumulator + 1;  // 增量累加
    }
    
    @Override
    public Long getResult(Long accumulator) {
        return accumulator;  // 直接返回累加结果
    }
    
    @Override
    public Long merge(Long a, Long b) {
        return a + b;  // 合并两个累加器
    }
}

4.3 增量关联：别 Join 全表

日志处理里经常要关联维表。比如把 IP 关联成城市，把 user_id 关联成用户名。

全量关联的做法是：把维表全量加载到内存，然后跟日志流做 Join。维表一大，内存就爆了。

增量关联的思路是：只加载新增或变更的维度数据。

我常用的三种增量关联方案：

• Bloom Filter 过滤——先判断 key 是否在维表里，不在的直接跳过，减少 Join 压力
• 增量维表加载——维表变更时，只更新变更的 key，不重新加载全表
• 异步 LRU 缓存——缓存最近访问的维度数据，过期自动淘汰

代码示例（增量维表关联）：

// 增量维表关联
public class IncrementalJoinProcessor {
    private Cache<String, DimensionData> dimCache;
    private BloomFilter<String> bloomFilter;
    
    public IncrementalJoinProcessor() {
        // 初始化 Bloom Filter（预计 1000 万条，误判率 1%）
        this.bloomFilter = BloomFilter.create(
            Funnels.stringFunnel(), 10_000_000, 0.01);
        
        // 初始化 LRU 缓存（最多 100 万条）
        this.dimCache = Caffeine.newBuilder()
            .maximumSize(1_000_000)
            .expireAfterWrite(1, TimeUnit.HOURS)
            .build();
    }
    
    public LogEvent enrich(LogEvent event) {
        String key = event.getUserId();
        
        // 1. Bloom Filter 快速过滤
        if (!bloomFilter.mightContain(key)) {
            return event;  // 维表里肯定没有，直接返回
        }
        
        // 2. 查缓存
        DimensionData dim = dimCache.getIfPresent(key);
        if (dim != null) {
            event.setCity(dim.getCity());
            return event;
        }
        
        // 3. 缓存未命中，查数据库（只查这一条）
        dim = queryDimensionFromDB(key);
        if (dim != null) {
            dimCache.put(key, dim);
            event.setCity(dim.getCity());
        }
        
        return event;
    }
    
    // 增量更新维表
    public void updateDimension(DimensionData dim) {
        dimCache.put(dim.getKey(), dim);
        bloomFilter.put(dim.getKey());
    }
}

4.4 增量去重：别存全量，存指纹

日志里重复数据很常见。网络重传、消费者重试、消息队列的 at-least-once 语义……

全量去重就是把所有见过的 ID 都存下来。数据量一大，存不下。

增量去重的核心是「空间换时间，但空间要省着换」。

我推荐三种增量去重方案：

方案	内存占用	准确率	适用场景
布隆过滤器	低（每条记录 10 bit 左右）	有误判（可配置）	允许少量重复的场景
HyperLogLog	极低（12KB 可统计亿级）	近似值	UV 统计等不需要精确值的场景
精确去重（Redis Set）	高（每条记录几十到几百字节）	100%	必须精确去重的场景

代码示例（布隆过滤器增量去重）：

// 布隆过滤器增量去重
public class DeduplicationProcessor {
    private BloomFilter<String> bloomFilter;
    private long totalProcessed = 0;
    private long duplicateCount = 0;
    
    public DeduplicationProcessor(long expectedInsertions) {
        // 预计插入 1 亿条，误判率 0.1%
        this.bloomFilter = BloomFilter.create(
            Funnels.stringFunnel(), 
            expectedInsertions, 
            0.001);
    }
    
    public boolean isDuplicate(String logId) {
        totalProcessed++;
        
        // 检查是否已经见过
        if (bloomFilter.mightContain(logId)) {
            duplicateCount++;
            return true;  // 可能是重复（有误判风险）
        }
        
        // 没见过，加入布隆过滤器
        bloomFilter.put(logId);
        return false;
    }
    
    public double getDuplicateRate() {
        return (double) duplicateCount / totalProcessed;
    }
}

4.5 四种技术的组合使用

实际项目中，这四种技术很少单独用。我见过一个典型的日志处理流水线：

1. 增量去重（布隆过滤器）——先过滤掉重复日志
2. 增量索引（段追加）——把去重后的日志写入搜索引擎
3. 增量聚合（滑动窗口）——实时统计各种指标
4. 增量关联（Bloom Filter + 缓存）——关联维度信息

你看，每个环节都在做「增量」，都在复用之前的结果。整个流水线的延迟从小时级降到了秒级。