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

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