如何使apache flink中的滑动窗口在达到窗口大小后才滑动？

7vux5j2d 于 2021-06-25 发布在 Flink

关注(0)|答案(2)|浏览(696)

专用数据流buysidevolumewma（数据流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呢？

apache-flink flink-streaming

来源：https://stackoverflow.com/questions/43558222/how-to-make-the-sliding-window-in-apache-flink-to-slide-only-after-the-window-si

2条答案

按热度按时间

eivgtgni1#

可以向窗口添加偏移。这是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>);

要了解更多信息：https://ci.apache.org/projects/flink/flink-docs-release-1.2/dev/windows.html

赞(0）回复(0）举报 2021-06-25

nhhxz33t2#

据我所知，在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])
      )
    }
  }

赞(0）回复(0）举报 2021-06-25

我来回答

如何使apache flink中的滑动窗口在达到窗口大小后才滑动？

2条答案

相关问题

热门标签

最新问答