参数为动态的 Spark 滞后函数

我需要在spark中实现lag函数；我可以像下面这样做 （使用 hive/temp Spark 表中的一些数据）

假设 DF 有这些行：

lagno:value
0, 100
0, 200
2, null
3, null

其中第一列是您要使用的实际滞后数，第二列是实际值。

当我运行此查询时，它有效：

DataFrame df;
DataFrame dfnew=df.select(
            org.apache.spark.sql.functions.lag( df.col("value"), 1 ).over(org.apache.spark.sql.expressions.Window.orderBy(new1.col("value"))));

这意味着如果对 lag no 的值进行硬编码，它就可以正常工作。

但是，如果我将滞后值作为参数传递，它就不起作用：

DataFrame dfnew=df.select(
            org.apache.spark.sql.functions.lag( df.col("value"),df.col("lagno").over(org.apache.spark.sql.expressions.Window.orderBy(new1.col("value"))));

我需要将 col 类型的参数强制转换为整数吗？

这不可能。窗口函数使用无法动态修改的固定大小的框架。你可以计算lag for 1..3然后选择当前行所需的一项。

CASE 
  WHEN lagno = 1 THEN LAG(value,  1) OVER w 
  WHEN lagno = 2 THEN LAG(value,  2) OVER w 
  ...
  ELSE value
END