如何使 Apache Flink 中的滑动窗口只有在达到窗口大小后才滑动？答案

【问题标题】：How to make the sliding window in Apache Flink to slide only after the window size is reached?如何使 Apache Flink 中的滑动窗口只有在达到窗口大小后才滑动？
【发布时间】：2017-04-22 10:39:57
【问题描述】：

私有数据流 buySideVolumeWMA(DataStream buyPressureTradeStream) {

    Integer windowSize = 3;
    Integer windowslide = 1;

    DataStream<Double> buySideVolumeWMAStream = buyPressureTradeStream.countWindowAll(windowSize, windowslide)
            .apply(new AllWindowFunction<String, Double, GlobalWindow>() {

                @Override
                public void apply(GlobalWindow window, Iterable<String> values, Collector<Double> out)
                        throws Exception {
                    Double buySideVolumeWMA = 0.0;
                    Integer weight = windowSize;
                    Integer numerator = 1;

                    for (String tradeString : values) {
                        JSONObject json = new JSONObject(tradeString);
                        Double tradeVolume = (Double) json.get("Volume");
                        buySideVolumeWMA += ((tradeVolume * numerator) / weight);
                        slf4jLogger.info("tradeVolume " + tradeVolume + " , " + "numerator , " + numerator
                                + " weight , " + weight + " buySideVolumeWMA " + buySideVolumeWMA);
                        numerator++;

                    }
                    numerator = 1;

                    out.collect(buySideVolumeWMA / 2);
                    buySideVolumePressure = buySideVolumeWMA / 2;
                    // slf4jLogger.info("buySideVolumePressure :" +
                    // buySideVolumePressure);


    buySideVolumeWMAStream.print().setParallelism(5);

    return buySideVolumeWMAStream;

}

================================================ ========================= 在这个程序中，我使用的窗口大小为 3，幻灯片大小为 1。我希望它在收到后开始滑动计数为 3 的流数据，然后仅开始滑动 1。但是发生的情况是，我的程序在接收到第一个数据时立即开始滑动，然后它接收到的每一个数据都滑动。那么如何让它只有在收到后才滑动计数 3 的数据然后滑动 1？

【问题讨论】：

您的问题不清楚。您能否详细说明您的问题是什么。你应该添加一个你得到什么和想要什么的例子。
@ImbaBalboa 。感谢您的回复。我在程序下方添加了有关我的问题的更多详细信息。您能指导我吗？谢谢

标签： apache-flink flink-streaming

【解决方案1】：

您可以向窗口添加偏移量。这是 Window 命令的第三个参数。在我看来，这样你可以稍后再开始。

文档中的示例：

// sliding processing-time windows offset by -8 hours
input
    .keyBy(<key selector>)
    .window(SlidingProcessingTimeWindows.of(Time.hours(12), Time.hours(1), Time.hours(-8)))
    .<windowed transformation>(<window function>);

【讨论】：

【解决方案2】：

据我所知，截至 2019 年 11 月和 Flink 1.9.1，这不是滑动窗口的功能。我知道这是因为窗口对象是独立的并且不共享任何状态。例如，如果使用键控流，则为每个窗口和键复制和存储窗口中的对象一次。

下面的过滤器保持足够的状态来忽略它收到的前 n 条消息。如果使用键控流（如.keyBy(...)），将为每个键保留一个单独的计数器，因为这是 Flink 管理 ValueState 对象的方式。

  /**
   * This filter suppresses the first n messages (inclusive of n). This behavior may be desired for use with sliding
   * windows when no output is desired until the full size of the window is reached.
   *
   * Example usage:
   * .filter(new SuppressFirstNFromSlidingWindow[(String, Int)](5))
   */
  class SuppressFirstNFromSlidingWindow[T](nToSuppress: Int) extends RichFilterFunction[T] {

    private var state_allowAll: ValueState[Boolean] = _
    private var state_numberSkipped: ValueState[Int] = _

    override def filter(value: T): Boolean = {

      if (state_allowAll.value()) return true

      val numberSkipped = state_numberSkipped.value()
      if (numberSkipped < nToSuppress) {
        state_numberSkipped.update(numberSkipped + 1)
        false
      } else {
        state_allowAll.update(true)
        true
      }
    }

    override def open(parameters: Configuration): Unit = {

      state_allowAll = getRuntimeContext.getState(
        new ValueStateDescriptor[Boolean]("allowAll", createTypeInformation[Boolean])
      )

      state_numberSkipped = getRuntimeContext.getState(
        new ValueStateDescriptor[Int]("numberSkipped", createTypeInformation[Int])
      )
    }
  }

【讨论】：