flink sql upsert模式写入mysql,es等key一定是groupby之后所有字段吗

classic Classic list List threaded Threaded
7 messages Options
Q
Reply | Threaded
Open this post in threaded view
|

flink sql upsert模式写入mysql,es等key一定是groupby之后所有字段吗

Q

举个例子比如我想要
insert into mysql_sink
select
     ID,
     amount,
     ………
 from source
groupby ID;

这里就是想按照id为key,在数据库里更新这个id对应的amount等其他值。

但这样子不能通过calcite的sqlvalidation,select后面不能有非聚合项,必须在groupby后面加上所有select后面的项。
但这样带来一个问题,这样子key的state无限增长(比如说amount是一个随机的double数),job跑不久就会fail掉。

请问大家有什么办法能只指定部分字段为key来规避掉这个问题,谢谢。
之前发过一封类似的邮件,没有回复,只好再发一封,见谅。




发自我的iPhone
Reply | Threaded
Open this post in threaded view
|

回复:flink sql upsert模式写入mysql,es等key一定是groupby之后所有字段吗

Yichao Yang
Hi


非活动key的状态是可清理的,可参考
https://blog.csdn.net/lp284558195/article/details/104609739
https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/streaming/query_configuration.html#:~:text=The%20minimum%20idle%20state%20retention,kept%20before%20it%20is%20removed.


Best,
Yichao Yang




------------------ 原始邮件 ------------------
发件人:&nbsp;"xu yihan"<[hidden email]&gt;;
发送时间:&nbsp;2020年6月4日(星期四) 下午5:59
收件人:&nbsp;"user-zh"<[hidden email]&gt;;

主题:&nbsp;flink sql upsert模式写入mysql,es等key一定是groupby之后所有字段吗




举个例子比如我想要
insert into mysql_sink
select
&nbsp;&nbsp;&nbsp;&nbsp; ID,
&nbsp;&nbsp;&nbsp;&nbsp; amount,
&nbsp;&nbsp;&nbsp;&nbsp; ………
&nbsp;from source
groupby ID;

这里就是想按照id为key,在数据库里更新这个id对应的amount等其他值。

但这样子不能通过calcite的sqlvalidation,select后面不能有非聚合项,必须在groupby后面加上所有select后面的项。
但这样带来一个问题,这样子key的state无限增长(比如说amount是一个随机的double数),job跑不久就会fail掉。

请问大家有什么办法能只指定部分字段为key来规避掉这个问题,谢谢。
之前发过一封类似的邮件,没有回复,只好再发一封,见谅。




发自我的iPhone
Reply | Threaded
Open this post in threaded view
|

Re: flink sql upsert模式写入mysql,es等key一定是groupby之后所有字段吗

Leonard Xu
In reply to this post by Q
Hi,

> 但这样子不能通过calcite的sqlvalidation,select后面不能有非聚合项,

select后费聚合值可以通过max()或sum()来取,因为已经按照key group by了,所以取出来的非聚合值只能有一条,

> 这样子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>
[2] https://issues.apache.org/jira/browse/FLINK-17829 <https://issues.apache.org/jira/browse/FLINK-17829>
Reply | Threaded
Open this post in threaded view
|

Re: flink sql upsert模式写入mysql,es等key一定是groupby之后所有字段吗

godfrey he
hi yihan,
如 Leonard 所说,你可以考虑使用 first_value, last_value 等聚合函数和赛选其他字段。
1.11开始支持ddl定义pk信息, 如果id在source表中也是pk字段,可以直接定义,
planner会利用该信息传递pk到sink表。

Bests,
Godfrey

Leonard Xu <[hidden email]> 于2020年6月4日周四 下午9:01写道:

> Hi,
>
> > 但这样子不能通过calcite的sqlvalidation,select后面不能有非聚合项,
>
> select后费聚合值可以通过max()或sum()来取,因为已经按照key group by了,所以取出来的非聚合值只能有一条,
>
> > 这样子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
> >
> [2] https://issues.apache.org/jira/browse/FLINK-17829 <
> https://issues.apache.org/jira/browse/FLINK-17829>
>
kcz
Reply | Threaded
Open this post in threaded view
|

回复:flink sql upsert模式写入mysql,es等key一定是groupby之后所有字段吗

kcz
In reply to this post by Leonard Xu
我大概get到你要说的需求,select那些其实是明细数据?但是没有跟聚合的数据拆开,所以才出现这种情况吧?





------------------ 原始邮件 ------------------
发件人: Leonard Xu <[hidden email]&gt;
发送时间: 2020年6月4日 21:01
收件人: user-zh <[hidden email]&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;
Reply | Threaded
Open this post in threaded view
|

Re: flink sql upsert模式写入mysql,es等key一定是groupby之后所有字段吗

x2009438
感谢各位,

我先试试用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 <[hidden email]> 写道:
>
> 我大概get到你要说的需求,select那些其实是明细数据?但是没有跟聚合的数据拆开,所以才出现这种情况吧?
>
>
>
>
>
> ------------------ 原始邮件 ------------------
> 发件人: Leonard Xu <[hidden email]&gt;
> 发送时间: 2020年6月4日 21:01
> 收件人: user-zh <[hidden email]&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;
Reply | Threaded
Open this post in threaded view
|

回复: flink sql upsert模式写入mysql,es等key一定是groupby之后所有字段吗

Yichao Yang
Hi


这里存在一个问题是,使用了last_value或者first_value这样的udaf,但是如果多条数据来到经过udaf处理后结果还是和之前一样的情况下,是不会产出回撤流数据的,可以观察下你是否需要考虑这种情况。


Best,
Yichao Yang




------------------&nbsp;原始邮件&nbsp;------------------
发件人:&nbsp;"x2009438"<[hidden email]&gt;;
发送时间:&nbsp;2020年6月5日(星期五) 上午9:06
收件人:&nbsp;"user-zh"<[hidden email]&gt;;

主题:&nbsp;Re: flink sql upsert模式写入mysql,es等key一定是groupby之后所有字段吗



感谢各位,

我先试试用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

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