Flink：进程窗口函数

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

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

对于一个只会触发一次的普通普通窗口，每个窗口的状态是无用的。只有当一个窗口以某种方式有多次触发（例如，延迟触发）时，您才能充分利用每个窗口的状态。如果您想从一个窗口到另一个窗口记住某些内容，那么您可以使用全局窗口状态来做到这一点。

有关使用每个窗口状态来记住要在后期触发中使用的数据的示例，请参阅 Flink 的 advanced window training 中的幻灯片 13-19。