基于Scala的Spark DataFrame多信号列差值计算优化实现问询

这个场景我之前也踩过坑，普通的lag函数确实搞不定带Null的情况——它只会死板地取窗口内的前一行，不管那行的值是不是Null。咱们得调整窗口逻辑，结合Spark支持的忽略Null的函数来实现，还能批量处理所有Signal列，不用重复写代码。

1. 核心思路拆解 #

要实现需求，关键要解决两个问题：

• 只在同一ID分组内按时间顺序找上一条非Null的Signal值
• 批量给所有Signal列生成对应的差值字段，Null行的差值也保持Null

2. 具体实现代码 #

第一步：定义正确的窗口 #

窗口需要按id分区（保证只在同一ID内计算），按date排序（保证时间顺序），同时允许函数访问当前行之前的所有行，用来找最近的非Null值：

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{col, last, when}

// 定义窗口：按id分区，按date排序，范围是从分区开头到当前行
val windowSpec = Window.partitionBy("id")
  .orderBy("date")
  .rowsBetween(Window.unboundedPreceding, Window.currentRow)

第二步：批量处理所有Signal列 #

先提取所有以signal开头的字段（如果你的列名规则不同，调整过滤逻辑即可），然后用foldLeft遍历每个字段，逐个添加差值列：

// 提取所有signal开头的列
val signalCols = df.columns.filter(_.startsWith("signal"))

// 批量生成差值列
val resultDf = signalCols.foldLeft(df) { (tempDf, colName) =>
  // 找到当前行之前最近的非null的signal值
  val lastNonNullValue = last(col(colName), ignoreNulls = true).over(windowSpec)
  // 计算差值：当前值非null时，用当前值减最近非null值；当前值为null时，差值也为null
  val diffCol = when(col(colName).isNotNull, col(colName) - lastNonNullValue)
    .otherwise(null)
    .alias(s"${colName}_diff")
  
  tempDf.withColumn(s"${colName}_diff", diffCol)
}

3. 逻辑验证（对应你的示例数据） #

拿id=051的signal01举例：

• 第一条记录值为1，lastNonNullValue就是自己，差值为0
• 第二条记录值为Null，差值保持Null
• 第三条记录值为3，lastNonNullValue会跳过第二条的Null，取第一条的1，差值为3-1=2，完全符合你的预期

再看id=052的signal03：

• 第一条值为0，差值0；第三条值为2，lastNonNullValue取第一条的0，差值2；第四条值为5，lastNonNullValue取第三条的2，差值3，完美匹配需求。

4. 为什么不用普通lag？ #

普通lag(colName,1,0)的问题在于：它只会取窗口内的前一行，哪怕那行是Null。比如id=051的第二条所有signal都是Null，用普通lag的话，第三条signal01的lag值会是Null，计算3-Null还是Null，根本不符合需求。而last(..., ignoreNulls=true)会自动跳过Null值，直接找到最近的有效记录，正好解决这个痛点。

内容的提问来源于stack exchange，提问作者Antony