参数为动态的火花滞后函数

Question

我需要在spark中实现滞后功能；我能够像下面那样做 （带有来自 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 类型的参数类型强制转换为整数？

Answer 1

这是不可能的。窗口函数使用无法动态修改的固定大小的帧。您可以为1..3 计算lag，然后选择当前行所需的一个。

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