Apache Spark-处理时态RDDs上的滑动窗口

Question

我最初的想法是简单地迭代，如果我们想要平均每X天，X次，每次只需按日期分组元素，并有一个偏移量。

所以如果我们有这样的场景：

天数：1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

第2组：（6,F）（7,G）（8,H）（9,I）（10,J）

第3组：（11,K）（12,L）（13,M）（14,N）（15,O）

第二次迭代：

Answer 1

如果您转换为DataFrame，这将变得简单得多--您只需将数据自联接到自身并找到平均值即可。假设我有这样一系列数据：

tsDF.show
date       amount
1970-01-01 10.0
1970-01-01 5.0
1970-01-01 7.0
1970-01-02 14.0
1970-01-02 13.9
1970-01-03 1.0
1970-01-03 5.0
1970-01-03 9.0
1970-01-04 9.0
1970-01-04 5.8
1970-01-04 2.8
1970-01-04 8.9
1970-01-05 8.1
1970-01-05 2.1
1970-01-05 2.78
1970-01-05 20.78

它卷起为：

tsDF.groupBy($"date").agg($"date", sum($"amount"), count($"date")).show
date       SUM(amount) COUNT(date)
1970-01-01 22.0        3
1970-01-02 27.9        2
1970-01-03 15.0        3
1970-01-04 26.5        4
1970-01-05 33.76       4

然后，我将需要创建一个UDF来为联接条件移位日期（注意，我使用偏移量=-2)，仅使用2天窗口：

def dateShift(myDate: java.sql.Date): java.sql.Date = {
  val offset = -2;
  val cal = Calendar.getInstance;
  cal.setTime(myDate);
  cal.add(Calendar.DATE, offset);
  new java.sql.Date(cal.getTime.getTime)
}
val udfDateShift = udf[java.sql.Date,java.sql.Date](dateShift)

val windowDF = tsDF.select($"date")
  .groupBy($"date")
  .agg($"date")
  .join(
    tsDF.select($"date" as "r_date", $"amount" as "r_amount"),
    $"r_date" > udfDateShift($"date") and $"r_date" <= $"date"
  )
  .groupBy($"date")
  .agg($"date",avg($"r_amount") as "2 day avg amount / record")

val windowDF.show
date       2 day avg amount / record
1970-01-01 7.333333333333333
1970-01-02 9.98
1970-01-03 8.58
1970-01-04 5.928571428571429
1970-01-05 7.5325