java

9.3 Flink 窗口聚合

eventd v3 实时指标计算的完整实现：用 Flink DataStream 并行三条窗口聚合（每分钟每用户订单量、近 5 分钟平均延迟、热点商品点击），用 ProcessWindowFunction 补齐窗口与维度信息后写入 Redis；同时说明增量聚合与全量聚合的选择、Redis 键设计为何必须带维度、迟到数据处理，以及本地执行环境下可重跑的断言方式。

Leeting Yan 6 分钟阅读 2865 字

实时面板需要哪些指标

eventd v3 的实时面板展示三类核心指标：

指标窗口类型维度刷新频率
每分钟订单量1 分钟滚动每用户1 分钟
近 5 分钟平均延迟5 分钟滑动全局1 分钟
热点商品点击1 分钟滚动每商品1 分钟

三个指标共用同一个 Kafka Source，但可以拆分为不同的 Flink 子作业或同作业内的多条流。

eventd-analytics 工程结构

eventd-v3/eventd-analytics/
├── pom.xml
├── src/main/java/com/plume/eventd/
│   ├── Event.java                      # 共享事件模型
│   ├── AnalyticsJob.java               # 主作业入口
│   ├── OrderCountAggregate.java        # 增量聚合函数
│   ├── AverageLatencyAggregate.java    # 延迟聚合
│   ├── HotProductAggregate.java        # 热点统计
│   ├── RedisMetricSink.java            # Redis Sink
│   └── RedisConfig.java                # Redis 连接配置
└── src/test/java/com/plume/eventd/
    └── AnalyticsJobTest.java           # MiniCluster 测试

主作业：并行三条窗口聚合

public class AnalyticsJob {

    // 三条分支统一输出这个记录：metric + 维度 + 窗口，缺一不可
    public record MetricRecord(
        String metric,          // order-count / avg-latency / hot-product
        String dimension,       // userId / global / symbol
        String dimensionValue,
        long windowStart,
        long windowEnd,
        String payload          // 聚合结果的 JSON
    ) {}

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        env.enableCheckpointing(30_000);
        env.setStateBackend(new EmbeddedRocksDBStateBackend());

        // Kafka Source
        KafkaSource<Event> source = KafkaSource.<Event>builder()
            .setBootstrapServers("kafka:9092")
            .setTopics("eventd-cleaned")
            .setGroupId("eventd-analytics")
            .setStartingOffsets(OffsetsInitializer.latest())
            .setValueOnlyDeserializer(new EventDeserializationSchema())
            .build();

        DataStream<Event> stream = env.fromSource(source,
            WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                .withTimestampAssigner((event, ts) -> event.eventTime()),
            "events"
        );

        // 分支 1：每分钟每用户订单量
        stream.filter(e -> e.type().equals("ORDER"))
            .keyBy(Event::userId)
            .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
            .aggregate(new OrderCountAggregate(), new ToMetricRecord<>("order-count"))
            .addSink(new RedisMetricSink());

        // 分支 2：近 5 分钟平均延迟（全局一个 key，用常量占位）
        stream.filter(e -> e.type().equals("ORDER"))
            .keyBy(e -> "global")
            .window(SlidingEventTimeWindows.of(Duration.ofMinutes(5), Duration.ofMinutes(1)))
            .aggregate(new AverageLatencyAggregate(), new ToMetricRecord<>("avg-latency"))
            .addSink(new RedisMetricSink());

        // 分支 3：热点商品（每分钟每商品订单金额汇总）
        stream.filter(e -> e.type().equals("ORDER") && e.symbol() != null)
            .keyBy(Event::symbol)
            .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
            .aggregate(new HotProductAggregate(), new ToMetricRecord<>("hot-product"))
            .addSink(new RedisMetricSink());

        env.execute("eventd-analytics");
    }
}

两处与旧写法不同的地方，都是 Flink 1.20 的要求：

  • 不再调用 setStreamTimeCharacteristic。Flink 1.12 起时间语义由 WatermarkStrategy 决定；这个方法已废弃，对窗口的时间语义没有影响，也不会改变自动 Watermark 间隔（那是 setAutoWatermarkInterval）。
  • 窗口用 java.time.Duration。Time.ofMinutes(1) 仍能编译，但 Time 自 1.12 起已废弃；of(Duration) 是当前签名。

聚合函数

OrderCountAggregate（订单计数）

public class OrderCountAggregate
    implements AggregateFunction<Event, Long, OrderCountResult> {

    @Override
    public Long createAccumulator() {
        return 0L;
    }

    @Override
    public Long add(Event event, Long accumulator) {
        return accumulator + 1;
    }

    @Override
    public OrderCountResult getResult(Long accumulator) {
        return new OrderCountResult(accumulator);
    }

    @Override
    public Long merge(Long a, Long b) {
        return a + b;
    }
}

public record OrderCountResult(long count) {}

AverageLatencyAggregate（平均延迟）

public class AverageLatencyAggregate
    implements AggregateFunction<Event, AverageAccumulator, LatencyResult> {

    private final Clock clock;   // Clock.SystemClock / FixedClock 都可序列化

    public AverageLatencyAggregate() {
        this(Clock.systemUTC());
    }

    public AverageLatencyAggregate(Clock clock) {
        this.clock = clock;
    }

    @Override
    public AverageAccumulator createAccumulator() {
        return new AverageAccumulator(0, 0);
    }

    @Override
    public AverageAccumulator add(Event event, AverageAccumulator acc) {
        long latency = clock.millis() - event.eventTime();
        return new AverageAccumulator(acc.sum + latency, acc.count + 1);
    }

    @Override
    public LatencyResult getResult(AverageAccumulator acc) {
        return new LatencyResult(
            acc.count > 0 ? acc.sum / acc.count : 0,
            acc.count
        );
    }

    @Override
    public AverageAccumulator merge(AverageAccumulator a, AverageAccumulator b) {
        return new AverageAccumulator(a.sum + b.sum, a.count + b.count);
    }
}

public record AverageAccumulator(long sum, long count) {}

public record LatencyResult(long avgLatencyMs, long sampleCount) {}

延迟用的是处理时间，不是事件时间：clock.millis() 是记录到达算子时的墙上时间。生产用 Clock.systemUTC()，测试注入 Clock.fixed(...)，否则同一个输入在不同时刻跑会得到不同延迟，断言无法稳定。用 Clock 而不是直接调 System.currentTimeMillis()，也是给测试留一个可控入口。

HotProductAggregate（热点商品）

public class HotProductAggregate
    implements AggregateFunction<Event, ProductAccumulator, HotProductResult> {

    @Override
    public ProductAccumulator createAccumulator() {
        return new ProductAccumulator(0, 0.0);
    }

    @Override
    public ProductAccumulator add(Event event, ProductAccumulator acc) {
        return new ProductAccumulator(acc.count + 1, acc.amount + event.amount());
    }

    @Override
    public HotProductResult getResult(ProductAccumulator acc) {
        return new HotProductResult(acc.count, acc.amount, acc.amount / acc.count);
    }

    @Override
    public ProductAccumulator merge(ProductAccumulator a, ProductAccumulator b) {
        return new ProductAccumulator(a.count + b.count, a.amount + b.amount);
    }
}

public record ProductAccumulator(int count, double amount) {}

public record HotProductResult(int orderCount, double totalAmount, double avgAmount) {}

用 ProcessWindowFunction 补齐窗口与维度

AggregateFunction 只能看到「值」，看不到窗口边界和 key。如果直接把它接到 Sink，OrderCountResult 里没有 userId、没有 windowStart，多条分支写同一个键会互相覆盖。

解决办法是 aggregate(aggregateFunction, processWindowFunction)：前者做增量聚合，后者在窗口触发时被调用一次，能拿到 key 和窗口：

public class ToMetricRecord<V> extends ProcessWindowFunction<
        V, AnalyticsJob.MetricRecord, String, TimeWindow> {

    private final String metric;

    public ToMetricRecord(String metric) {
        this.metric = metric;
    }

    @Override
    public void process(String key, Context ctx, Iterable<V> values,
                        Collector<AnalyticsJob.MetricRecord> out) {
        // 增量聚合与 ProcessWindowFunction 组合时，Iterable 里只有一个元素
        V result = values.iterator().next();
        out.collect(new AnalyticsJob.MetricRecord(
            metric,
            dimensionOf(metric),
            key,
            ctx.window().getStart(),
            ctx.window().getEnd(),
            JsonMapper.toJson(result)
        ));
    }

    private String dimensionOf(String metric) {
        return switch (metric) {
            case "order-count" -> "userId";
            case "hot-product" -> "symbol";
            default -> "global";
        };
    }
}

代价是每个窗口多一次函数调用，但换来的是可查询、可断言的键。这是 Sink 之前必须做的转换，不是可选的优化。

Redis Sink 实现

public class RedisMetricSink extends RichSinkFunction<AnalyticsJob.MetricRecord> {
    private transient RedisClient client;
    private transient StatefulRedisConnection<String, String> connection;
    private transient RedisCommands<String, String> commands;

    @Override
    public void open(OpenContext parameters) {
        client = RedisClient.create("redis://localhost:6379");
        connection = client.connect();
        commands = connection.sync();
    }

    @Override
    public void invoke(MetricRecord record, Context context) {
        // key 必须带维度，否则同一窗口的多个用户会互相覆盖
        String key = String.format("eventd:metrics:%s:%d:%s",
            record.metric(), record.windowStart(), record.dimensionValue());
        commands.setex(key, 300, record.payload());

        // 同时写入最新值，供 SSE 首帧快速返回；它只是缓存，不能用于验收
        commands.set("eventd:metrics:" + record.metric() + ":latest", record.payload());

        // 发布到面板订阅的频道。10.x 的 SSE live 流监听的就是这个频道；
        // Pub/Sub 不持久化，断线期间的消息由上面的 latest 快照补齐
        commands.publish("eventd:metrics", record.payload());
    }

    @Override
    public void close() {
        if (connection != null) connection.close();
        if (client != null) client.shutdown();
    }
}

四点说明：

  • 必须继承 RichSinkFunction，不能只 implements SinkFunction。open(OpenContext) 和 close() 来自 RichFunction，只实现 SinkFunction 时这两个方法没有可覆盖的父方法，@Override 无法编译。
  • 这里用 sync() 同步写，是为了让「写失败」直接以异常抛出，便于故障排查。代价是每条记录一次往返；批量与异步写法、以及通常更高的吞吐，见 7.2 侧输出连接器 。
  • :latest 的覆盖写在重放时可能被旧窗口「写回去」，它不是幂等的。要保留这个指针就得加窗口起点比较，做法见 11.1 端到端 Exactly-Once 。
  • publish 是给面板 SSE 的实时通道，不是状态的一部分：Pub/Sub 不落盘、不重放，订阅者断线期间发出去的消息就没了。所以先写 setex/:latest 再 publish，重连时面板能靠快照补齐当前值。

地址写死 localhost:6379 只适合本机调试。12.2 的 compose 里 Redis 的服务名是 redis，作业跑在容器内时 localhost 指向容器自己；生产应从参数或环境变量读取地址，部署细节见 12.2 部署与运维 。

SinkFunction 在 Flink 1.20 已标记废弃，推荐用 Sink V2；1.20 仍完整支持 RichSinkFunction，迁移方式见 7.3 自定义连接器 。

Redis 数据结构：

keyvalueTTL
eventd:metrics:order-count:1700004600000:alice{"count":12}300s
eventd:metrics:order-count:latest最新一条窗口数据无
eventd:metrics:avg-latency:1700004600000:global{"avgLatencyMs":230,"sampleCount":45}300s
eventd:metrics:hot-product:1700004600000:BTC{"orderCount":8,"totalAmount":800.0,"avgAmount":100.0}300s

端到端测试

// v3 的 Event 是 12 个组件的 record。测试里用一个小工厂只写关心的字段，
// 其余取默认值（version=1、source=eventd-v2、status=0）
private static Event event(String id, String type, String userId,
                           double amount, String symbol, long ts) {
    return new Event(id, type, 1, "eventd-v2", ts, userId, amount, symbol,
            0, "127.0.0.1", "web", Map.of());
}

// 用「第 n 秒」构造时间戳，让窗口边界一眼可见
private static long instantAt(long seconds) {
    return seconds * 1000;
}

@Test
void testAnalyticsJob() throws Exception {
    // 并行度 1，结果顺序可预测；多并行度时下面按集合断言，不按下标
    StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(1);

    // 测试数据
    DataStream<Event> testStream = env.fromCollection(Arrays.asList(
        event("e1", "ORDER", "alice", 100.0, "BTC", instantAt(0)),
        event("e2", "ORDER", "alice", 200.0, "BTC", instantAt(30)),
        event("e3", "ORDER", "bob",   300.0, "ETH", instantAt(45)),
        event("e4", "ORDER", "alice", 150.0, "BTC", instantAt(65)),  // 进入下一个窗口
        event("e5", "REFUND", "alice", 50.0, "BTC", instantAt(20))   // 被 filter 过滤
    )).assignTimestampsAndWatermarks(
        WatermarkStrategy.<Event>forMonotonousTimestamps()
            .withTimestampAssigner((e, ts) -> e.eventTime())
    );

    // 收集订单计数结果（CollectSink 用静态集合，原因见 8.3）
    CollectSink.VALUES.clear();
    testStream.filter(e -> e.type().equals("ORDER"))
        .keyBy(Event::userId)
        .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
        .aggregate(new OrderCountAggregate())
        .addSink(new CollectSink<>());

    env.execute("Analytics Test");

    // alice[0,60) = 2，alice[60,120) = 1，bob[0,60) = 1；e5 被过滤
    // 有界源结束时 Flink 会发出 Long.MAX_VALUE 的 Watermark，所有窗口都会触发
    List<Long> counts = CollectSink.<OrderCountResult>values().stream()
        .map(OrderCountResult::count)
        .sorted()
        .toList();
    assertEquals(List.of(1L, 1L, 2L), counts);
}

两个容易写错的断言点：

  • 不要用 results.get(0)。bob 也会产生一条结果，且算子输出顺序不保证；按集合或排序后比较。
  • 不要假设末尾窗口不会触发。fromCollection 是有界源，作业结束时会发出最大 Watermark，[60,120) 这个窗口同样会触发。若要测试「窗口未触发」，必须用无界源（如 TestHarness 手动推 Watermark），不能靠「数据不够」。CollectSink 的正确写法见 8.3 测试 。

测试要点

场景验证内容
窗口边界事件正好在窗口边界时的归属
乱序数据Watermark 与 allowedLateness 的行为
迟到数据sideOutputLateData 的输出
空窗口无数据时窗口不触发
keyBy 分布不同 key 的数据是否独立计算

小结

聚合窗口输出
订单量1min 滚动OrderCountResult
平均延迟5min 滑动LatencyResult
热点商品1min 滚动HotProductResult
Redis Sink无窗口键含窗口起点与维度，TTL 5 分钟

第 9 章结束。下一节进入告警与面板：10.1 实时指标计算 。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「java」更多文章

  1. 15.3 全书总结与学习路径回顾
  2. 15.2 响应式流处理的演进趋势
  3. 15.1 引擎选型指南