用于n天滚动计数的星火斯卡拉联非洲新议程

Question

我正试图计算过去30天中与当前事务时间戳有关的密钥的数量。我知道我可以直接使用一个窗口来计算这个值，但是对于这个窗口中的一个键(分区)，我有大量的数据，因此在我的硬件上这项工作永远不会完成。

我看到一些博客谈论联非新议程。有人能建议我怎么做吗，指给我看好文章(找不到适合用例的东西)。类似窗口+联非新议程

下面是我的window rollingWindow = Window .partitionBy(keyColumn) .orderBy(unix_timestamp) .rangeBetween(-30_24_3600，currentRow)

我不能用任何其他键来划分数据，因为那样的话计数就会出错。如有任何建议，不胜感激。

Answer 1

用UDAFs实现一个切片窗口可能有点棘手。如果问题是分区非常大，那么您可能也会遇到同样的问题，因为您也需要为密钥进行聚合。

我会首先检查幻灯片窗口的问题，为什么要花这么长时间才能完成(它完成了吗？)在选择其他选择之前。您必须考虑到Spark优化提高了内存/CPU/存储效率，因此，使用UDAF (先验地)处理JVM类型的数据不会提高性能。

如果分区太大，可以尝试使用RDDs和mapPartitions中的迭代器转换。迭代器是一种逐个访问元素集合的方法。迭代器不会将内存中所有元素的全部集合一起加载。相反，迭代器一个接一个地加载元素“。

您可以使用迭代器与mapPartitions进行迭代器转换。这里的要点是，您应该首先从这些分区构建一个RDD，并应用函数mapPartition：

// AN example of pair RDD
val rdd : RDD[(Int, Array[Double])] = ???

rdd.mapPartitions { iterator : Iterator[(Int, Array[Double])] =>
  // Read the elements one by one with the next method
  // 
  iterator.map(el => (el, windowFunction(buffer))
}

小心，如果您在一个集合中加载迭代器(如List )，您将遇到相同的内存问题。因此，您需要使用迭代器的map函数来执行聚合函数，并在另一个迭代器中转换这个迭代器。另一个问题是，按日期范围执行函数可能会更困难，因为您只能读取迭代器一次，因此需要处理这个问题。