如何在 Flink 中为 WindowFunction 将 fold() 转换为 AggregateFunction？答案

【问题标题】：How to convert fold() to an AggregateFunction in Flink for WindowFunction?如何在 Flink 中为 WindowFunction 将 fold() 转换为 AggregateFunction？
【发布时间】：2019-05-14 18:04:39
【问题描述】：

我试图通过删除已弃用的类将旧的雅虎流式 Flink 基准版本转换为新版本。

我现在坚持将弃用的 fold() 转换为 aggregate()。我无法将 fold 的现有参数映射到汇总的参数。

//old version using fold
 val windowedCounts = windowedEvents.fold(new WindowedCount(null, "", 0, new java.sql.Timestamp(0L)),
          (acc: WindowedCount, r: (String, String, Timestamp)) => {
            val lastUpdate = if (acc.lastUpdate.getTime < r._3.getTime) r._3 else acc.lastUpdate
            acc.count += 1
            acc.lastUpdate = lastUpdate
            acc
          },
          (key: Tuple, window: TimeWindow, input: Iterable[WindowedCount], out: Collector[WindowedCount]) => {
            val windowedCount = input.iterator.next()
            println(windowedCount.lastUpdate)
            out.collect(new WindowedCount(new java.sql.Timestamp(window.getStart), key.getField(0), windowedCount.count, windowedCount.lastUpdate))
            //out.collect(new WindowedCount(new java.sql.Timestamp(window.getStart), key.getField(0), windowedCount.count, windowedCount.lastUpdate))
          }
        )

val windowedCounts = windowedEvents.aggregate(new CountAggregate)

我想通过扩展 AggregateFunction 类（类似）来创建一个 CountAggregate 类：

class CountAggregate extends AggregateFunction[(String, String, Timestamp), WindowedCount, Collector[WindowedCount]] {
    override def createAccumulator() = WindowedCount(null, "", 0, new java.sql.Timestamp(0L))

    override def accumulate(acc: WindowedCount, r: (String, String, Timestamp)): WindowedCount = {
      val lastUpdate = if (acc.lastUpdate.getTime < r._3.getTime) r._3 else acc.lastUpdate
      acc.count += 1
      acc.lastUpdate = lastUpdate
      acc
          }

    override def getValue (acc: WindowedCount)  = { (key: Tuple, window: TimeWindow, input: Iterable[WindowedCount], out: Collector[WindowedCount]) =>
      val windowedCount = input.iterator.next()
      println(windowedCount.lastUpdate)
      out.collect(new WindowedCount(new java.sql.Timestamp(window.getStart), key.getField(0), windowedCount.count, windowedCount.lastUpdate))
    }

我们将不胜感激重写 CountAggregate 类的任何帮助。

【问题讨论】：

标签： scala user-defined-functions apache-flink flink-streaming fold

【解决方案1】：

您需要指定AggregateFunction 和ProcessWindowFunction 来执行最后的getValue 步骤：

val windowedCounts = windowedEvents.aggregate(
      new CountAggregate(),
      new WindowAggregateFunction())

class CountAggregate extends AggregateFunction[(String, String, Timestamp), WindowedCount, WindowedCount] {
  override def createAccumulator() = WindowedCount(null, "", 0, new java.sql.Timestamp(0L))

  override def add(value: (String, String, Timestamp), acc: WindowedCount): WindowedCount = {
    val lastUpdate = if (acc.lastUpdate.getTime < value._3.getTime) value._3 else acc.lastUpdate
    WindowedCount(null, "", acc.count + 1, lastUpdate)
  }

  override def getResult(accumulator: WindowedCount): WindowedCount = {
    accumulator
  }

  override def merge(a: WindowedCount, b: WindowedCount): WindowedCount = {
    WindowedCount(null, "", a.count + b.count, if (a.lastUpdate.getTime < b.lastUpdate.getTime) b.lastUpdate else a.lastUpdate)
  }
}

class WindowAggregateFunction extends ProcessWindowFunction[WindowedCount, WindowedCount, Tuple, TimeWindow]() {
  override def process(key: Tuple, context: Context, elements: Iterable[WindowedCount], out: Collector[WindowedCount]): Unit = {
    val windowedCount = elements.iterator.next()
    out.collect(WindowedCount(new java.sql.Timestamp(context.window.getStart), key.getField(0), windowedCount.count, windowedCount.lastUpdate))
  }
}

【讨论】：