apache-spark - 大量窗口函数导致内存不足(滞后、超前)

Question

我需要使用多个领先和滞后从数据集中计算附加功能。大量超前和滞后会导致内存不足错误。

数据框:

|----------+----------------+---------+---------+-----+---------|
| DeviceID | Timestamp      | Sensor1 | Sensor2 | ... | Sensor9 |
|----------+----------------+---------+---------+-----+---------|
|          |                |         |         |     |         |
| Long     | Unix timestamp | Double  | Double  |     | Double  |
|          |                |         |         |     |         |
|----------+----------------+---------+---------+-----+---------|

窗口定义:

// Each window contains about 600 rows
val w = Window.partitionBy("DeviceID").orderBy("Timestamp") 

计算额外特征:

var res = df
val sensors = (1 to 9).map(i => s"Sensor$i")

for (i <- 1 to 5) {
  for (s <- sensors) {
    res = res.withColumn(lag(s, i).over(w))
         .withColumn(lead(s, i)).over(w)
  }

  // Compute features from all the lag's and lead's
  [...]
}

系统信息:

RAM: 16G
JVM heap: 11G

该代码在小数据集上给出了正确的结果，但在输入数据为 10GB 时给出了内存不足错误。
我认为罪魁祸首是大量的窗口函数，因为 DAG 显示了很长的序列

Window -> WholeStageCodeGen -> Window -> WholeStageCodeGen ...

无论如何以更有效的方式计算相同的特征？
例如，是否有可能获得 lag(Sensor1, 1), lag(Sensor2, 1), ..., lag(Sensor9, 1) 而不调用 lag(..., 1) 九次？

如果上一个问题的答案是否定的，那么如何避免内存不足？我已经尝试增加分区数。

Answer 1

你可以尝试类似的东西

res = res.select('*', lag(s"Sensor$1", 1).over(w), lag(s"Sensor$1", 2).over(w), ...)

也就是说，将所有内容写入 select而不是很多 withColumn
那么计划中将只有 1 个窗口。也许它有助于性能。