你好,

如果我沒有理解錯你的應用場景的話,你想達成的結果應該是類似這篇討論 [1] 裡提到的問題對吧?
從最新的 flink 文檔 [2] 中來看應該無法透過你期望的 on duplicate key 語句來實現,
或許可以嘗試在 SELECT 語句上達成,舉例來說你可以在原有的 select 語句之外多添加 group by,如下:

insert into t select a, last_value(b ignore nulls) as b, last_value(c
> ignore nulls) as c from $(original_select_statement) group by a;


不過目前 last_value 似乎不支持 ignore nulls,你可以考慮自己實現一個 UDAF 來達成。
另外,這樣的做法也會造成 flink state 不斷增長 (由於 group by 的緣故),所以需要多加小心,比如適當的配置 state ttl。

best regards,

[1]
https://stackoverflow.com/questions/48144641/mysql-using-on-duplicate-key-update-coalesce
[2]
https://ci.apache.org/projects/flink/flink-docs-master/docs/dev/table/sql/insert/#syntax

Ye Chen <chenye...@163.com> 於 2021年8月2日 週一 下午4:08寫道:

> 你好,我们用的1.11版本。
>
> 需求:table t 有三个字段(a,b,c)
> 我们的场景只想根据主键a更新部分字段b,其余的字段c原有的值保持不变,
>
> 例如mysql 支持:
> insert into t(a,b) select 1,2 on duplicate key update b=2;
> 主键重复的时候只更新字段b,字段c的值不变。
> 但是flink sql 目前只支持全字段更新:insert into t(a,b,c) select 1,2,3 。
> 我在sql-client测试了一下:insert into t(a,b) select 1,2 on duplicate key update
> b=2;  会报错 不支持 on duplicate key update
> 同时也测试了一下:insert into t(a,b) select 1,2  也会报错,字段数量不匹配;
> [ERROR] Could not execute SQL statement. Reason:
> org.apache.flink.table.api.ValidationException: Column types of query
> result and sink for registered table 'default_catalog.default_database.t'
> do not match.
> Cause: Different number of columns.
> 我查看了https://issues.apache.org/jira/browse/FLINK-18726
> 使用最新版的1.13  sql-client测试了一下 insert into t(a,b) select 1,2
> ,相比1.11版本报错,1.13可以执行成功.但是发现和文档中描述的一样,字段c会插入null。如果原有的一条数据是1,2,3执行sql后会变成1,2,null,会造成字段c的数据丢失,这是不允许的。
> 请问这种根据主键更新部分字段的场景 使用flink sql应该怎么处理?
>
> 在 2021-08-02 15:39:09,"silence" <slle...@aliyun.com.INVALID> 写道:
> >用的什么版本这个应该已经在1.13中修复了https://issues.apache.org/jira/browse/FLINK-18726
> >不行的话可以在ddl中限制列的数量
> >
> >
> >------------------------------------------------------------------
> >发件人:Ye Chen <chenye...@163.com>
> >发送时间:2021年8月2日(星期一) 11:37
> >收件人:user-zh <user-zh@flink.apache.org>; silence <slle...@aliyun.com>
> >主 题:Re:回复:场景题:Flink SQL 不支持 INSERT INTO… ON DUPLICATE KEY UPDATE ?
> >
> >你好,我试了一下,如果表的ddl是三个字段,但是insert只指定两个字段的话,会报错:
> >[ERROR] Could not execute SQL statement. Reason:
> >org.apache.flink.table.api.ValidationException: Column types of query
> result and sink for registered table 'default_catalog.default_database.t'
> do not match.
> >Cause: Different number of columns.
> >我们的需求是想根据主键更新部分字段
> >-------------------------------------
> >需求:现有table
> >CREATE TABLE t (
> > a        bigint,
> > b        bigint,
> > c        bigint,
> >  PRIMARY KEY (a) NOT ENFORCED
> >) WITH (
> >...
> >);
> >我们的场景只想根据主键a更新部分字段b,其余的字段c保持不变,
> >例如mysql 支持   insert into t(a,b,c) select '1','2','3' on duplicate key
> update b='4';主键重复的时候只更新字段b,字段c的值不变。
> >我在官方文档中没找到这个用法,sql-client也测试了一下也不支持 on duplicate key update,会报错。
> >请问这种根据主键更新部分字段的场景 使用flink sql应该怎么处理?
> >
> >
> >在 2021-08-02 10:47:55,"silence" <slle...@aliyun.com.INVALID> 写道:
> >>如果只想更新部分字段的话可以试下
> >>insert into t(a,b) select a,b from xxxxx
> >>
> >>
> >>------------------------------------------------------------------
> >>发件人:Ye Chen <chenye...@163.com>
> >>发送时间:2021年7月30日(星期五) 17:57
> >>收件人:user-zh <user-zh@flink.apache.org>
> >>主 题:场景题:Flink SQL 不支持 INSERT INTO… ON DUPLICATE KEY UPDATE ?
> >>
> >>现有table
> >>CREATE TABLE t (
> >> a        bigint,
> >> b        bigint,
> >> c        bigint,
> >>  PRIMARY KEY (a) NOT ENFORCED
> >>) WITH (
> >>...
> >>);
> >>
> >>
> >>我们的场景只想根据主键a更新部分字段b,其余的字段保持不变,例如
> >>mysql 支持   insert into t(a,b,c) select '1','2','3' on duplicate key
> update b='4';
> >>主键重复的时候只更新字段b,字段c的值不变
> >>
> >>
> >>我在官方文档中没找到这个用法,sql-client也测试了一下也不支持 on duplicate key update,会报错。
> >>请问这种部分字段更新的场景 使用flink sql应该怎么处理?
> >>
> >>
> >
> >
> >
>

回复