Spark RDD：填充不规则时间序列

Question

所以我有一个带有不规则时间序列数据的 RDD：

1,
4、
6、
..等等

我需要将其填充到常规时间序列中：

1,
2、
3、
4、
5、
6、
..等等

到目前为止，我已经创建了一个具有 1、2、3、4、5、6、.. 的 RDD，然后将其左外连接到原始 RDD，这给了我：

1,
2、
3、
4、
5、
6、
..等等

所以我面临的问题是用前一个非 Null 行中的值填充那些 2,3,5。

我宁愿在 RDD 级别上执行此操作而不使用 sparkSQL，这当然是最后的选择。转到 scala 数组级别并不是很吸引人，因为对于性能问题，我更愿意将其保持在 RDD 级别。

谢谢

Answer 1

没有初始join 的相对简单的解决方案。让我们从虚拟数据和辅助函数开始：

val rdd = sc.parallelize(Seq(
    (3L, 1.0), (1L, 4.0), (5L, 3.6), (7L, 0.2), (8L, 0.0)))

def fillTimePoints(xs: Array[(Long, Double)]) = xs match {
  case Array((xTime, xValue), (yTime, _)) => {
    val diff = yTime - xTime

    if (diff == 0) Seq((xTime, xValue))
    else (xTime, xValue) +: (1 until diff.toInt)
      .map(_.toLong)
      .map(i => (i + xTime, xValue))
  }

  case _ => Seq.empty[(Long, Double)]
}

现在剩下的就是在排序的 RDD 上滑动：

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

rdd.sortBy(_._1).sliding(2).flatMap(fillTimePoints).collect

//  Array[(Long, Double)] = Array((1,4.0), (2,4.0), (3,1.0), 
//    (4,1.0), (5,3.6), (6,3.6), (7,0.2))

注意事项：

• sliding 是开发人员 API 的一部分。其类中的大多数方法在最近的版本中已被弃用。仍然可以从头开始编写代码，但现在它应该可以工作了，

• 您可能更喜欢使用RangePartitioner 后跟repartitionAndSortWithinPartitions 而不是排序。然后您可以使用mapPartitions 应用局部滑动，并将preservePartitioning 设置为true，最后填补空白（再次preservePartitioning）。它需要更多的工作，但您会得到使用RangePartitioner 分区的输出，这可能很有用。