apache-spark - 参数为动态的 Spark 滞后函数

Question

我需要在spark中实现滞后功能；我可以像下面那样做(使用来自配置单元/临时 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"))));

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

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

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