Flink：如何将已弃用的fold转换为aggregate？

Question

我正在按照Flink的快速入门示例监控维基百科编辑流进行操作。

这个示例是用Java编写的，而我正在使用Scala进行实现，如下：

/**
 * Wikipedia Edit Monitoring
 */
object WikipediaEditMonitoring {
  def main(args: Array[String]) {
    // set up the execution environment
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val edits: DataStream[WikipediaEditEvent] = env.addSource(new WikipediaEditsSource)

    val result = edits.keyBy( _.getUser )
      .timeWindow(Time.seconds(5))
      .fold(("", 0L)) {
        (acc: (String, Long), event: WikipediaEditEvent) => {
          (event.getUser, acc._2 + event.getByteDiff)
        }
      }

    result.print

    // execute program
    env.execute("Wikipedia Edit Monitoring")
  }
}

然而，Flink中的fold函数已经弃用，建议使用aggregate函数。

但我没有找到有关如何将弃用的 fold 转换为 aggregrate 的示例或教程。

有什么想法吗？可能不仅仅是应用 aggregrate。

更新

我有另一个实现如下：

/**
 * Wikipedia Edit Monitoring
 */
object WikipediaEditMonitoring {
  def main(args: Array[String]) {
    // set up the execution environment
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val edits: DataStream[WikipediaEditEvent] = env.addSource(new WikipediaEditsSource)

    val result = edits
      .map( e => UserWithEdits(e.getUser, e.getByteDiff) )
      .keyBy( "user" )
      .timeWindow(Time.seconds(5))
      .sum("edits")

    result.print

    // execute program
    env.execute("Wikipedia Edit Monitoring")
  }

  /** Data type for words with count */
  case class UserWithEdits(user: String, edits: Long)
}

我也想知道如何使用自定义 AggregateFunction 进行实现。 更新 我遵循了这个文档：AggregateFunction，但是有以下问题：

在版本 1.3 的接口 AggregateFunction 的源代码中，您将看到 add 确实返回 void：

void add(IN value, ACC accumulator);

但是对于版本1.4 AggregateFunction，它返回：

ACC add(IN value, ACC accumulator);

我该如何处理这个问题？

我正在使用的Flink版本是1.3.2，但该版本的文档中没有AggregateFunction，而artifactory中尚未发布1.4版本。

- fluency03

Answer 1

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer08
import org.apache.flink.streaming.connectors.wikiedits.{WikipediaEditEvent, WikipediaEditsSource}

class SumAggregate extends AggregateFunction[WikipediaEditEvent, (String, Int), (String, Int)] {
  override def createAccumulator() = ("", 0)

  override def add(value: WikipediaEditEvent, accumulator: (String, Int)) = (value.getUser, value.getByteDiff + accumulator._2)

  override def getResult(accumulator: (String, Int)) = accumulator

  override def merge(a: (String, Int), b: (String, Int)) = (a._1, a._2 + b._2)
}

object WikipediaAnalysis extends App {
  val see: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  val edits: DataStream[WikipediaEditEvent] = see.addSource(new WikipediaEditsSource())

  val result: DataStream[(String, Int)] = edits
    .keyBy(_.getUser)
    .timeWindow(Time.seconds(5))
    .aggregate(new SumAggregate)
//    .fold(("", 0))((acc, event) => (event.getUser, acc._2 + event.getByteDiff))
  result.print()

  result.map(_.toString()).addSink(new FlinkKafkaProducer08[String]("localhost:9092", "wiki-result", new SimpleStringSchema()))
  see.execute("Wikipedia User Edit Volume")
}

- Casel Chen

---

欢迎来到Java的世界。 - Joan

Answer 2

您可以在Flink 1.4文档中找到有关AggregateFunction的一些文档，其中包括一个示例。

1.3.2版本仅限于与可变累加器类型一起使用，其中添加操作修改累加器。这已经在Flink 1.4中得到了修复，但尚未发布。

- David Anderson