Flink 1.14窗口：如何在新事件到来时立即输出中间结果？

使用Flink 1.14 + Java，结合Table API与DataStream API（toDataStream/toAppendStream），需求如下：

• 从Kafka读取事件数据
• 按小时维度执行sum、count等聚合操作
• 新事件到来时立即将聚合结果Upsert到Cassandra（新增或更新主键对应的sum、count值）

当前遇到的问题：采用TUMBLE窗口SQL实现时，任务仅在窗口过期（每小时）后才向Cassandra发送结果，但Flink官方明确窗口聚合仅输出最终结果，无法实时输出中间更新值。

解决方案 #

方法1：Table API 无窗口持续聚合（推荐） #

放弃TUMBLE窗口，改用基于处理时间/事件时间的小时截断分组配合Upsert流输出，每一条事件到来都会实时更新对应小时的聚合结果，直接输出到Cassandra。

示例代码

// 初始化流表环境
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

// 注册Kafka源表
tableEnv.executeSql("CREATE TABLE kafka_events (" +
    "  event_id STRING," +
    "  value INT," +
    "  proc_time AS PROCTIME()" + // 定义处理时间属性
    ") WITH (" +
    "  'connector' = 'kafka'," +
    "  'topic' = 'your_topic'," +
    "  'properties.bootstrap.servers' = 'localhost:9092'," +
    "  'format' = 'json'" +
    ")");

// 执行持续聚合查询：按小时截断处理时间分组
Table resultTable = tableEnv.sqlQuery("SELECT " +
    "  DATE_TRUNC('HOUR', proc_time) AS hour_window," + // 按小时截断作为分组主键
    "  COUNT(*) AS event_count," +
    "  SUM(value) AS total_value" +
    " FROM kafka_events" +
    " GROUP BY DATE_TRUNC('HOUR', proc_time)");

// 转换为Upsert流（处理聚合结果的更新/新增）
DataStream<Row> upsertStream = tableEnv.toChangelogStream(resultTable)
    .filter(row -> row.getKind() != RowKind.DELETE) // 过滤删除操作（按需选择）
    .map(row -> row);

// 下沉到Cassandra（实现Upsert逻辑）
CassandraSink.addSink(upsertStream)
    .setQuery("INSERT INTO hourly_stats (hour_window, event_count, total_value) VALUES (?, ?, ?) IF NOT EXISTS; " +
              "UPDATE hourly_stats SET event_count=?, total_value=? WHERE hour_window=?;")
    .setHost("localhost")
    .build();

关键说明

• DATE_TRUNC('HOUR', proc_time)将处理时间截断到小时级别，替代TUMBLE窗口作为聚合分组键，属于持续聚合模式
• toChangelogStream输出包含INSERT/UPDATE_AFTER类型的数据流，完美匹配Cassandra的Upsert需求
• 每一条事件都会触发对应小时分组的聚合计算，并立即输出更新后的结果，无需等待窗口关闭

方法2：DataStream API 用KeyedProcessFunction维护状态 #

如果更倾向于DataStream API的灵活性，可以用KeyedProcessFunction维护每个小时的聚合状态，实时更新并输出结果。

示例代码

// 从Kafka读取事件流
DataStream<Event> eventStream = env.addSource(new FlinkKafkaConsumer<>(
    "your_topic",
    new EventDeserializationSchema(),
    kafkaProps
));

// 按小时截断后的时间戳分组
KeyedStream<Event, String> keyedStream = eventStream
    .keyBy(event -> {
        // 将当前处理时间截断到小时，生成分组键
        long hourTimestamp = (System.currentTimeMillis() / 3600000) * 3600000;
        return String.valueOf(hourTimestamp);
    });

// 自定义ProcessFunction维护聚合状态
DataStream<HourlyStats> resultStream = keyedStream.process(new KeyedProcessFunction<String, Event, HourlyStats>() {
    // 存储当前小时的聚合状态
    private ValueState<HourlyStats> hourlyStatsState;

    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<HourlyStats> stateDesc = new ValueStateDescriptor<>(
            "hourly-stats",
            TypeInformation.of(HourlyStats.class)
        );
        hourlyStatsState = getRuntimeContext().getState(stateDesc);
    }

    @Override
    public void processElement(Event event, Context ctx, Collector<HourlyStats> out) throws Exception {
        HourlyStats currentStats = hourlyStatsState.value();
        if (currentStats == null) {
            // 初始化新小时的聚合状态
            long hourTs = (ctx.timerService().currentProcessingTime() / 3600000) * 3600000;
            currentStats = new HourlyStats(hourTs, 0, 0);
        }
        // 更新聚合值
        currentStats.setEventCount(currentStats.getEventCount() + 1);
        currentStats.setTotalValue(currentStats.getTotalValue() + event.getValue());
        hourlyStatsState.update(currentStats);
        // 立即输出更新后的结果
        out.collect(currentStats);

        // 注册定时器：当前小时结束后清理状态，避免内存泄漏
        long nextHourTs = ((ctx.timerService().currentProcessingTime() / 3600000) + 1) * 3600000;
        ctx.timerService().registerProcessingTimeTimer(nextHourTs);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<HourlyStats> out) throws Exception {
        // 小时过期后清理对应状态
        hourlyStatsState.clear();
    }
});

// 下沉到Cassandra实现Upsert
CassandraSink.addSink(resultStream)
    .setQuery("INSERT INTO hourly_stats (hour_window, event_count, total_value) VALUES (?, ?, ?) IF NOT EXISTS; " +
              "UPDATE hourly_stats SET event_count=?, total_value=? WHERE hour_window=?;")
    .setHost("localhost")
    .build();

关键说明

• 以小时截断后的时间戳作为分组键，每个键对应一个小时的聚合状态
• 每处理一条事件就更新状态并立即输出结果，满足实时Upsert需求
• 通过定时器在小时过期后清理状态，避免状态无限堆积

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