你好,请问下,my_parse是个udf吧 然后有没有什么操作可以使用udtf解析出多个字段 , 这些字段直接就是source表的字段 , 然后选出时间字段定义 watermark , 类似如下 CREATE TABLE sourceTable( request_uri STRING, (column_1,column_2,heart_time) as udtf_parse(request_uri) )with(......); 哈哈,不知道有没有这样的语法
在 2020-06-24 12:24:46，"Jark Wu" <[email protected]> 写道： >你可以在 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 >> > >>
