org.apache.flink.table.api.ValidationException: Expected LocalReferenceExpression. Got: timestamp_expected localreferenceexpression. got: duration

情景：在设置滑动窗口时遇到的问题：Expected LocalReferenceExpression. Got: timestamp，详细代码如下：

Table table = tableEnv.fromDataStream(stream);
table
.window(Slide.over(lit(1).hours()).every(lit(5).minutes()).on(($("timestamp").rowtime()).as("ts")).as("w"))
.groupBy($("w"), $("itemId"))
.select($("itemId"), $("itemId").count().as("itemCount"), $("w").end().as("windowEnd"))
.window(Over.partitionBy($("windowEnd")).orderBy($("itemCount").desc()).as("row_num"))
.select($("itemId"), $("itemCount"), $("windowEnd"))
.where($("row_num").between(0, 3));
1
2
3
4
5
6
7
8

原因是：需要在定义表设置事件时间，在表API中不能设置事件时间。
原来代码修改为：

Table table = tableEnv.fromDataStream(stream,$("itemId"),$("timestamp").rowtime().as("ts"));
Table tableResult =
       table
       .window(Slide.over(lit(1).hours())
               .every(lit(5).minutes())
               .on($("ts"))
               .as("w"))
       .groupBy($("w"), $("itemId"))
       .select($("itemId"), $("itemId").count().as("itemCount"), $("w").end().as("windowEnd"))
       .window(Over.partitionBy($("windowEnd"))
               .orderBy($("itemCount").desc())
               .as("rank"))
       .select($("itemId"), $("itemCount"), $("windowEnd"))
       .where($("row_num").between(0, 3));
1
2
3
4
5
6
7
8
9
10
11
12
13
14