在Flink的聚合原语中具有与HOP_START等价的

Question

我正试图用Flink SQL在一个跳跃窗口上做一个指数衰减移动平均。我需要访问窗口的一个边框，在下面的HOP_START:

    SELECT                                                                              
      lb_index one_key,
    -- I have access to this one:
      HOP_START(proctime, INTERVAL '0.05' SECOND, INTERVAL '5' SECOND) start_time,
    -- Aggregation primitive:
      SUM(
        Y * EXP(TIMESTAMPDIFF(
          SECOND, 
          proctime, 
    -- This one throws:
          HOP_START(proctime, INTERVAL '0.05' SECOND, INTERVAL '5' SECOND)
      )))
    FROM write_position                                                                
    GROUP BY lb_index, HOP(proctime, INTERVAL '0.05' SECOND, INTERVAL '5' SECOND)

我得到了以下堆栈跟踪：

11:55:37.011 [main] DEBUG o.a.c.p.RelOptPlanner - For final plan, using Aggregate(groupBy: (lb_index), window: (SlidingGroupWindow('w$, 'proctime, 5000.millis, 50.millis)), select: (lb_index, SUM($f2) AS Y, start('w$) AS w$start, end('w$) AS w$end, proctime('w$) AS w$proctime))
11:55:37.011 [main] DEBUG o.a.c.p.RelOptPlanner - For final plan, using Calc(select: (lb_index, proctime, *(payload.Y, EXP(/(CAST(/INT(Reinterpret(-(HOP_START(PROCTIME(proctime), 50, 5000), PROCTIME(proctime))), 1000)), 1000))) AS $f2))
11:55:37.011 [main] DEBUG o.a.c.p.RelOptPlanner - For final plan, using rel#459:DataStreamScan.DATASTREAM.true.Acc(table=[_DataStreamTable_0])
Exception in thread "main" org.apache.flink.table.codegen.CodeGenException: Unsupported call: HOP_START 
If you think this function should be supported, you can create an issue and start a discussion for it.
    at org.apache.flink.table.codegen.CodeGenerator$$anonfun$visitCall$3.apply(CodeGenerator.scala:1027)
    at org.apache.flink.table.codegen.CodeGenerator$$anonfun$visitCall$3.apply(CodeGenerator.scala:1027)
    at scala.Option.getOrElse(Option.scala:121)
    at org.apache.flink.table.codegen.CodeGenerator.visitCall(CodeGenerator.scala:1027)
    at org.apache.flink.table.codegen.CodeGenerator.visitCall(CodeGenerator.scala:66)

它确实说它在聚合和之外工作时未实现。这就是为什么我认为这是一个范围问题。

现在，问题是：我可以转换这个表达式并在聚合之外进行最后的处理，如exp（x，y）=exp（x）*exp（y）；但我坚持使用TIMESTAMPDIFF（这在我上一期中创造了奇迹）。我还没有找到将时间属性转换为数字类型的方法；此外，我不喜欢对UNIX时间戳进行幂运算，即使我缩小了它们。

无论如何，这项工作会有点笨拙，可能会有另一种方式。我不知道如何在不抛出的情况下，将此SQL片段中的作用域推送到窗口作用域中并具有开始时间。

Answer 1

我建议你尝试使用HOP_PROCTIME（）而不是HOP_START（）。这里解释了这些差异，但效果将是您将拥有一个proctime属性而不是时间戳，我希望这将使TIMESTAMPDIFF满意。