带有先前值列表的 Spark Streaming RDD

Question

我不太确定如何使用 Spark Streaming 来做到这一点，这真的让我很头疼。 我有一些用户的输入流，例如 User1、User2 等带有一些参数，这些参数现在并不重要。然后我对这些参数做一些处理，得到用户的当前状态，比如A，B，C，像这样：

   (User1, A)
   (User2, B)
   (User3, B)

在不久的将来，我会收到一组新的事件，可能是这样的：

   (User1, C)
   (User2, B)
   (User3, A)

到现在都没有问题。棘手的部分是如何为最后 X 秒内的所有事件生成一个可能具有这样结构的 RDD：

   (User1, (A, C))
   (User2, (B, B))
   (User3, (B, A))

如果不是很明显，我想生成用户在最后 X 秒内处于的先前状态的列表。有没有办法通过 Spark Streaming 实现这一点？到目前为止，我通常会使用 Esper 来做这件事，但 Spark Streaming 看起来仍然对这项工作很有用。

在上面的例子中，我并没有将自己限制在一个元组中，括号只是为了解释这个过程。

编辑：

这是我到目前为止的一个代码示例。

val userZone = positions.map(position =>
      if (
        position.latitude > 45.445 &&
          position.latitude < 45.48 &&
          position.longitude > 9.14 &&
          position.longitude < 9.24)
        (position, "center")
      else (position, "out")

    val groupedByZones = userZone.groupByKeyAndWindow(Seconds(120))
    groupedByZones.foreachRDD(rdd =>
    {
      rdd.foreach(record => {
        println(record._1.userid + ":" + record._2)
      })
    })

所以我想要的输出应该是这样的：

U1: (center, out, center)
U2 (center,center,out)...

但现在我只得到单独的值，例如：

U1 : center
U1 : out
U1 : center

我希望我是清楚的。

Answer 1

所以在 Spark Streaming 中有几种方法可以做到这一点。想到的最简单的是 groupByKeyAndWindow（您可以在 http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.streaming.dstream.PairDStreamFunctions 中看到）。