你可以在 DDL 中直接用计算列去从 request_uri 里获得 heart_time 哈，然后在这个计算列上定义 watermark 即可。 例如:
CREATE TABLE sourceTable ( request_uri STRING, heart_time AS my_parse(request_uri), WATERMARK FOR heart_time AS heart_time - INTERVAL '1' SECOND ) WITH ( ....... ); 虽然这会导致重复解析两遍。 Best, Jark On Wed, 24 Jun 2020 at 12:09, Weixubin <[email protected]> wrote: > > > > 感谢你提供了子查询的思路，不过经过试验后有点可惜，这似乎还是满足不了我们的需求。 > > > 我们的场景是从阿里云SLS读取消息。每条消息有一个字段是request_uri。 > 第一个数据处理过程就是将 request_uri 解析为多个属性（多列），存成一行，作为一条记录。 > 第二个数据处理过程就是将每条记录里的 声明heart_time为事件时间属性并使用5秒延迟水印策略，进行开窗聚合处理。 > > > //如果应用到子查询的话,Flink是不支持这样做的。WATERMARK FOR 水印声明只能在DDL里应用。如下： > select > ..ts as TO_TIMESTAMP(heart_time,'yyyy-MM-ddHH:mm:ss') , > WATERMARK FOR ts AS ts - INTERVAL '1' SECOND > from sourceTable , LATERAL TABLE(ParseUriRow(request_uri)) as T( )….· > > > //如果应用到source，则一开始并不知道heart_time 的值 > CREATE TABLE sourceTable ( > request_uri STRING > ..ts as TO_TIMESTAMP(heart_time,'yyyy-MM-ddHH:mm:ss') , > WATERMARK FOR ts AS ts - INTERVAL '1' SECOND > ) WITH ( ....... ); > > > 只能等待Flink 1.11 尝试是否可以用View作为中间临时表，并对View进行 WATERMARK水印声明 > Thanks > Bin > > 在 2020-06-23 15:28:50，"Leonard Xu" <[email protected]> 写道： > >Hi > >我的意思是你如果中间结果表如果要输出，那你就一个sink写到中间结果表（什么表根据你的需要），一个sink写到你的最终结果表，在这两个sink之前的`select > * from sourceTable , LATERAL TABLE(ParseUriRow(request_uri)) as T( )….` > 的这段sql是可以复用的，就和 VIEW的作用类似。 > > > >如果你不需要中间结果表，只是要最终的结果表，那你写个嵌套的sql就行了，里层是`select * from sourceTable , > LATERAL TABLE(ParseUriRow(request_uri)) as T( )….·，外层是 group by, > 插入最终的结果表就能满足需求了吧。 > > > >祝好， > >Leonard Xu > > > > > >> 在 2020年6月23日，15:21，Weixubin <[email protected]> 写道： > >> > >> > >> > >> > >> Hi， > >> 关于这句 “把 ` select * from sourceTable , LATERAL > TABLE(ParseUriRow(request_uri)) as T( )….` 这段 sql insert 到中间结果表 和 > group后再写入最终结果表就可以了” > >> 我不太明白所说的中间结果表是什么意思， 我所理解为在数据库创建多一张middleSinkTable表，作为中间结果表。请问这样理解是否有误？ > 可否简单举个例子。 > >> Thanks, > >> Bin > >> > >> > >> > >> > >> > >> > >> > >> > >> > >> > >> 在 2020-06-23 11:57:28，"Leonard Xu" <[email protected]> 写道： > >>> Hi, > >>> 是的，这个是在 FLINK 1.11 的版本里支持的，当前 FLINK 1.11代码已经冻结，正在测试中，最近应该快发布了，所以master > 分支上的版本号为1.12-SNAPSHOT > >>> ，等1.11版本发布了就可以看到对应的文档。 > >>> > >>> 回到这个问题，flink sql(blink planner) 支持 multiple-sinks 的， 在 1.10 > 版本也可以一个作业搞定。 把 ` select * from sourceTable , LATERAL > TABLE(ParseUriRow(request_uri)) as T( )….` 这段 sql insert 到中间结果表 和 > group后再写入最终结果表就可以了。效果和使用 VIEW 应该类似，因为 planner 会做分段优化。 > >>> 另外建议使用1.10.1版本，1.10.1在1.10的基础上修复了不少bug，也可以等1.11发布了试用下1.11。 > >>> > >>> > >>> 祝好， > >>> Leonard Xu > > >
