多时间窗口唯一元素计数：Process函数还是增量聚合方案？

问题描述 #

我们需要统计输入流在多个滑动时间窗口内的唯一ele3数量：

• 输入数据结构：InputData(ele1: Integer, ele2: String, ele3: String)
• 分组规则：按ele1和ele2分组
• 窗口配置：滑动间隔15分钟，窗口大小分别为1小时、12小时、24小时，结果每15分钟刷新一次
• 原方案：使用ProcessWindowFunction存储所有事件，窗口结束后统计唯一元素，内存消耗高
• 尝试优化方案：结合ReduceFunction与ProcessWindowFunction增量聚合，用HashSet存储唯一元素，但内存消耗反而上升，堆转储显示大量HashMap实例（HashSet底层）占用内存超过原方案

当前环境：Flink 1.14.2，Yarn部署，状态后端为HashMap。

问题分析 #

当前Reduce方案内存更高的核心原因：

1. 对象重复创建：每次reduce调用都会生成新的UserData对象，额外增加了对象实例的内存开销
2. HashSet的底层开销：HashSet基于HashMap实现，每个元素对应一个哈希表节点，相比直接存储原始InputData对象，会产生更多的内存 overhead
3. 状态存储效率低：HashMap状态后端将所有状态保存在堆内存中，大窗口（12/24小时）的大量唯一元素会直接耗尽堆内存

优化方案 #

1. 改用AggregateFunction实现增量聚合 #

AggregateFunction是Flink中更高效的增量聚合接口，直接操作累加器，避免不必要的对象创建。示例代码如下：

精确计数实现

// 输入数据类
public class InputData {
    private Integer ele1;
    private String ele2;
    private String ele3;
    // getter、constructor省略
}

// 结果输出类
public class Metrics {
    private Integer ele1;
    private String ele2;
    private int uniqueEle3Count;
    private long windowEnd;
    // getter、constructor省略
}

// AggregateFunction实现唯一元素计数
public class UniqueEle3Aggregate implements AggregateFunction<InputData, HashSet<String>, Integer> {
    @Override
    public HashSet<String> createAccumulator() {
        return new HashSet<>();
    }

    @Override
    public HashSet<String> add(InputData value, HashSet<String> accumulator) {
        accumulator.add(value.getEle3());
        return accumulator;
    }

    @Override
    public Integer getResult(HashSet<String> accumulator) {
        return accumulator.size();
    }

    @Override
    public HashSet<String> merge(HashSet<String> a, HashSet<String> b) {
        a.addAll(b);
        return a;
    }
}

// 结合ProcessWindowFunction获取窗口时间（可选）
public class WindowMetricsProcess extends ProcessWindowFunction<Integer, Metrics, Tuple2<Integer, String>, TimeWindow> {
    @Override
    public void process(Tuple2<Integer, String> key, Context context, Iterable<Integer> counts, Collector<Metrics> out) {
        int count = counts.iterator().next();
        out.collect(new Metrics(key.f0, key.f1, count, context.window().getEnd()));
    }
}

// 业务流处理
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<InputData> inputStream = ...;

inputStream
    .keyBy(input -> Tuple2.of(input.getEle1(), input.getEle2()))
    // 1小时窗口，15分钟滑动
    .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(15)))
    .aggregate(new UniqueEle3Aggregate(), new WindowMetricsProcess())
    .addSink(...);

2. 切换到RocksDB状态后端 #

HashMap状态后端完全依赖堆内存，对于大窗口场景，改用RocksDB状态后端可以将状态持久化到磁盘，大幅降低堆内存压力：

// 配置RocksDB状态后端
StateBackend stateBackend = new EmbeddedRocksDBStateBackend(true);
env.setStateBackend(stateBackend);

// 配置检查点（确保状态持久化）
env.enableCheckpointing(Time.minutes(10).toMillis(), CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setCheckpointStorage("hdfs:///path/to/checkpoints");

3. 近似计数：使用布隆过滤器（Bloom Filter） #

如果业务允许一定的误判率（比如1%-5%），布隆过滤器可以大幅降低内存消耗——它不需要存储实际元素，仅用位向量记录元素存在性：

import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;

public class BloomFilterUniqueAggregate implements AggregateFunction<InputData, BloomFilter<String>, Integer> {
    // 误判率设置为1%
    private static final double FALSE_POSITIVE_RATE = 0.01;

    @Override
    public BloomFilter<String> createAccumulator() {
        // 预估每个窗口的元素数量，可根据业务调整
        return BloomFilter.create(Funnels.stringFunnel(StandardCharsets.UTF_8), 100000, FALSE_POSITIVE_RATE);
    }

    @Override
    public BloomFilter<String> add(InputData value, BloomFilter<String> accumulator) {
        accumulator.put(value.getEle3());
        return accumulator;
    }

    @Override
    public Integer getResult(BloomFilter<String> accumulator) {
        // 返回近似计数
        return (int) accumulator.approximateElementCount();
    }

    @Override
    public BloomFilter<String> merge(BloomFilter<String> a, BloomFilter<String> b) {
        // 注意：Guava布隆过滤器不支持merge，若需合并可使用Apache Commons的BloomFilter实现
        throw new UnsupportedOperationException("Merge not supported for Guava BloomFilter");
    }
}

4. 优化状态清理 #

确保窗口过期后及时清理状态，避免无效状态占用内存：

// 配置状态TTL
StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.hours(24)) // 设置TTL为窗口最大时长+滑动间隔，确保窗口过期后状态被清理
    .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .build();

// 将TTL配置应用到全局状态
env.getConfig().setStateTtlConfig(ttlConfig);

方案选择建议 #

• 精确计数场景：优先使用AggregateFunction+HashSet+RocksDB状态后端，既保证增量聚合的效率，又缓解内存压力
• 近似计数场景：使用布隆过滤器，内存消耗仅为HashSet的1/10甚至更低
• 避免使用ReduceFunction+ProcessWindowFunction的组合，其对象创建开销和状态存储效率均不如AggregateFunction

内容的提问来源于stack exchange，提问作者Sachin Ravadi