使用 Flink 获取键控窗口中的计数

Question

我通过 Scala 接口使用 Flink 进行一些数据处理。我有一些元组形式的用户数据：

(user1, "titanic")
(user1, "titanic")
(user1, "batman")
(user2, "star wars")
(user2, "star wars")
(user2, "batman")

我想按用户键入，创建一个窗口，然后计算用户在该窗口中观看特定电影的次数，以便我最终得到从每部电影到每个用户的观看次数的地图。例如，对于user1，正确的输出是Map("titanic" -> 2, "batman" -> 1)。 我知道我的代码的第一部分应该看起来像这样：

keyedStream.keyBy(0).window(EventTimeSessionWindows.withGap(Time.minutes(10)))

但是我不知道如何在窗口中进行进一步的聚合，以便最终得到每个用户/窗口的视图计数地图。我尝试编写自己的 AggregateFunction 将这些计数收集到可变映射中，但不幸的是可变映射不可序列化，因此失败了。

我该怎么做？

Answer 1

您应该能够使用AggregateFunction来解决问题：

source
.keyBy(0)
.timeWindow(Time.seconds(10L))
.aggregate(new AggregateFunction[(String, String), (String, Map[String, Int]), (String, Map[String, Int])] {
override def createAccumulator(): (String, Map[String, Int]) = ("", Map())
override def add(value: (String, String), accumulator: (String, Map[String, Int])): (String, Map[String, Int]) = {
val counter = accumulator._2.getOrElse(value._2, 0)
(value._1, accumulator._2 + (value._2 -> (counter + 1)))
}
override def getResult(accumulator: (String, Map[String, Int])): (String, Map[String, Int]) = accumulator
override def merge(a: (String, Map[String, Int]), b: (String, Map[String, Int])): (String, Map[String, Int]) = {
(a._1, (a._2.keySet ++ b._2.keySet) map (k => k -> (a._2.getOrElse(k, 0) + b._2.getOrElse(k, 0))) toMap)
}
})