Flink： ProcessWindow函数

Question

我最近在Flink的新版本中学习ProcessWindowFunction。它表示ProcessWindowFunction支持全局状态和窗口状态。我使用Scala API进行了尝试。到目前为止，我可以让全球状态运转起来，但我没有任何运气让它成为窗口状态。我要做的是处理系统日志，并统计由主机名和严重性级别键入的日志数。我想计算两个相邻窗口之间日志计数的差异。下面是我实现ProcessWindowFunction的代码。

class LogProcWindowFunction extends ProcessWindowFunction[LogEvent, LogEvent, Tuple, TimeWindow] {
  // Create a descriptor for ValueState
  private final val valueStateWindowDesc = new ValueStateDescriptor[Long](
    "windowCounters",
    createTypeInformation[Long])

  private final val reducingStateGlobalDesc = new ReducingStateDescriptor[Long](
    "globalCounters",
    new SumReduceFunction(),
    createTypeInformation[Long])

  override def process(key: Tuple, context: Context, elements: Iterable[LogEvent], out: Collector[LogEvent]): Unit = {
    // Initialize the per-key and per-window ValueState
    val valueWindowState = context.windowState.getState(valueStateWindowDesc)
    val reducingGlobalState = context.globalState.getReducingState(reducingStateGlobalDesc)
    val latestWindowCount = valueWindowState.value()
    println(s"lastWindowCount: $latestWindowCount ......")
    val latestGlobalCount = if (reducingGlobalState.get() == null) 0L else reducingGlobalState.get()
    // Compute the necessary statistics and determine if we should launch an alarm
    val eventCount = elements.size
    // Update the related state
    valueWindowState.update(eventCount.toLong)
    reducingGlobalState.add(eventCount.toLong)
    for (elem <- elements) {
      out.collect(elem)
    }
  }
}

我总是从窗口状态中得到0值，而不是以前应该更新的计数。我已经为这个问题挣扎了几天。有人能帮我弄清楚吗？谢谢。

Answer 1

每窗口状态的范围是一个窗口实例。对于上面的过程方法，每次调用它时，一个新窗口都在范围内，因此latestWindowCount始终为零。

对于一个普通的、普通的、只会触发一次的窗口，每个窗口的状态都是无用的。只有当一个窗口有多次触发（例如延迟触发）时，才能充分利用每个窗口的状态。如果您试图从一个窗口记住下一个窗口中的内容，那么可以使用全局窗口状态来执行此操作。

有关使用每个窗口状态来记住在后期点火中使用的数据的示例，请参阅Flink高级窗口培训中的幻灯片13-19。