我试图通过删除不推荐使用的类，将旧的雅虎flink流媒体基准版本转换为新版本。
我现在正忙于将不推荐使用的fold（）转换为aggregate（）。我无法将fold的现有参数Map到聚合参数。

//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（新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类都将不胜感激。