可能有不同的选择，但其中之一是实施 WindowFunction 然后跑 apply 开窗后：

eventsStream
    .keyBy(extractKey)
    .timeWindow(Time.minutes(2))
    .apply(new MyWindowFunction)

( WindowFuntion 获取输入值类型、输出值类型和键类型的类型参数。）
这里有一个例子。让我复制一下相关的片段：

/**User-defined WindowFunction to compute the average temperature of SensorReadings */
class TemperatureAverager extends WindowFunction[SensorReading, SensorReading, String, TimeWindow] {
  /**apply() is invoked once for each window */
  override def apply(
    sensorId: String,
    window: TimeWindow,
    vals: Iterable[SensorReading],
    out: Collector[SensorReading]): Unit = {
    // compute the average temperature
    val (cnt, sum) = vals.foldLeft((0, 0.0))((c, r) => (c._1 + 1, c._2 + r.temperature))
    val avgTemp = sum / cnt
    // emit a SensorReading with the average temperature
    out.collect(SensorReading(sensorId, window.getEnd, avgTemp))
}

我不知道你的数据看起来如何，所以我不能尝试一个完整的答案，但这应该作为灵感。