感谢各位,

我先试试用Last_value这样的aggregate function绕过去。

@kcz
可能我表达不很清楚,具体描述一下遇到的具体场景就是:收到原始的数据,去mysql或者es里做维表关联,然后再以upsert的模式将结果写回mysql或es。

举个例子来说,我想按id为key更新整行数据(比如还有个字段amount是个随机的double类型值)
select
id,
amount,
……
groupby id;

这样子不行,必须
select
id,
amount,
…
groupby id,amount,……;

这样子,假设id是固定有限个,那么state个数应该也是有限的,但是加上amount,我理解state的状态就会+♾了。
也就是说我实际并不需要分组聚合,保存状态,用groupby只是为了让connector识别到keyfields。


发自我的iPhone

> 在 2020年6月5日,08:46,kcz <573693...@qq.com> 写道:
> 
> 我大概get到你要说的需求,select那些其实是明细数据?但是没有跟聚合的数据拆开,所以才出现这种情况吧?
> 
> 
> 
> 
> 
> ------------------ 原始邮件 ------------------
> 发件人: Leonard Xu <xbjt...@gmail.com&gt;
> 发送时间: 2020年6月4日 21:01
> 收件人: user-zh <user-zh@flink.apache.org&gt;
> 主题: 回复:flink sql upsert模式写入mysql,es等key一定是groupby之后所有字段吗
> 
> 
> 
> Hi,
> 
> &gt; 但这样子不能通过calcite的sqlvalidation,select后面不能有非聚合项,
> 
> select后费聚合值可以通过max()或sum()来取,因为已经按照key group by了,所以取出来的非聚合值只能有一条,
> 
> &gt; 这样子key的state无限增长(比如说amount是一个随机的double数),job跑不久就会fail掉。
> 
> State 可以配置ttl的,过期清理参考[1]
> 
> 另外,即将发布的1.11中,支持在jdbc table 上定义primary key, 不用强制要求写upsert 的query,文档正在撰写中[2]
> 
> Best,
> Leonard Xu
> 
> 
> [1] 
> https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/streaming/query_configuration.html#idle-state-retention-time
>  
> <https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/streaming/query_configuration.html#idle-state-retention-time&gt;
> [2] https://issues.apache.org/jira/browse/FLINK-17829 
> <https://issues.apache.org/jira/browse/FLINK-17829&gt;

回复