Flink SQL: Casting time中跳跃窗口上的指数衰减移动平均值

Question

现在我们在 Flink 中拥有了带有花哨窗口的 SQL，我正在尝试从他们的 SQL 路线图/预览版 2017-03 帖子中引用衰减的移动平均线“，以”未来 Flink 版本对 Table API 和 SQL 的发布将会发生什么“：

table
  .window(Slide over 1.hour every 1.second as 'w)
  .groupBy('productId, 'w)
  .select(
    'w.end,
    'productId,
    ('unitPrice * ('rowtime - 'w.start).exp() / 1.hour).sum / (('rowtime - 'w.start).exp() / 1.hour).sum)

这是我的尝试（也受到方解石腐烂的例子的启发）：

SELECT                                                                              
  lb_index one_key,                                                           
  HOP_START(proctime, INTERVAL '0.05' SECOND, INTERVAL '5' SECOND) start_time,  
  SUM(Y * 
      EXP(
        proctime - 
        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)

时间是处理时间，通过从AppendStream表创建write_position，我们得到处理时间，如下所示:

tEnv.registerTable(
    "write_position", 
    tEnv.fromDataStream(appendStream, "lb_index, Y, proctime.proctime"))

我收到此错误：

Cannot apply '-' to arguments of type '<TIME ATTRIBUTE(PROCTIME)> - <TIME ATTRIBUTE(PROCTIME)>'. 
Supported form(s): '<NUMERIC> - <NUMERIC>' '<DATETIME_INTERVAL> - <DATETIME_INTERVAL>' '<DATETIME> - <DATETIME_INTERVAL>'

我已经尝试过将程序时间转换为我所知道的所有其他类型（试图到达NUMERIC应许之地），但我就是找不到如何使它工作。

我错过了什么吗？时间是否是您无法转换的非常特殊的“系统更改数字”时间？如果是这样，仍然必须有某种方法将其与HOP_START（proctime,...）值进行比较。

Answer 1

您可以使用timestampDiff减去两个时间点（参见文档）。你这样使用它

TIMESTAMPDIFF(timepointunit, timepoint1, timepoint2)

其中时间点单位可以是秒、分钟、小时、天、月或年。

我还没有尝试过处理时间，但它确实适用于事件时间字段，所以希望它会。