Spark 发现时间戳中的空白答案

【问题标题】：Spark finding gaps in timestampsSpark 发现时间戳中的空白
【发布时间】：2016-10-09 19:33:20
【问题描述】：

我有一个 Pair RDD，它由 (Key, (Timestamp,Value)) 条目组成。

读取数据的时候，entry是按照timestamp排序的，所以RDD的每个partition都要按照timestamp排序。我想要做的是，为每个键找到 2 个连续时间戳之间的最大差距。

我现在考虑这个问题很久了，鉴于 sparks 提供的功能，我不知道如何实现。我看到的问题是：我在做一个简单的地图时丢失了订单信息，所以这是不可能的。在我看来，groupByKey 失败也是因为特定键的条目太多，尝试这样做会给我一个java.io.IOException: No space left on device

任何有关如何解决此问题的帮助都会非常有帮助。

【问题讨论】：

按键和时间戳排序。然后是数据的线性传递。
我可能不会在 Spark 中这样做。本质上是线性传递的东西不太适合 Spark。您可能可以转换为 DF 并使用 windows，但我从未这样做过。 cran.r-project.org/web/packages/dplyr/vignettes/… 看起来很相关
@TheArchetypalPaul 你能给我指出一个更合适的流行技术吗

标签： algorithm scala apache-spark

【解决方案1】：

按照The Archetypal Paul 的建议，您可以使用DataFrame 和窗口函数。首先需要导入：

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.lag

下一个数据必须转换为DataFrame：

val df = rdd.mapValues(_._1).toDF("key", "timestamp")

为了能够使用lag 函数，我们需要一个窗口定义：

val keyTimestampWindow = Window.partitionBy("key").orderBy("timestamp")

可用于选择：

val withGap = df.withColumn(
  "gap", $"timestamp" - lag("timestamp", 1).over(keyTimestampWindow)
)

最后是groupBy 和max：

withGap.groupBy("key").max("gap")

按照The Archetypal Paul 的第二条建议，您可以按键和时间戳排序。

val sorted = rdd.mapValues(_._1).sortBy(identity)

使用这样排列的数据，您可以通过滑动和按键减少找到每个键的最大间隙：

import org.apache.spark.mllib.rdd.RDDFunctions._

sorted.sliding(2).collect {
  case Array((key1, val1), (key2, val2)) if key1 == key2 => (key1, val2 - val1)
}.reduceByKey(Math.max(_, _))

先重新分区和排序的相同想法的另一种变体：

val partitionedAndSorted = rdd
  .mapValues(_._1)
  .repartitionAndSortWithinPartitions(
    new org.apache.spark.HashPartitioner(rdd.partitions.size)
  )

这样的数据可以转换

val lagged = partitionedAndSorted.mapPartitions(_.sliding(2).collect {
  case Seq((key1, val1), (key2, val2)) if key1 == key2 => (key1, val2 - val1)
}, preservesPartitioning=true)

和reduceByKey:

lagged.reduceByKey(Math.max(_, _))

【讨论】：