> 有其他办法吗 可以尝试group by id并配合UDF：LAST_VALUE， SQL示例如下： insert into sink_es select id,LAST_VALUE(column_name) from binlog group by id;
Best, Oliver yunchang > 2020年5月7日 下午8:55，Luan Cooper <[email protected]> 写道： > > 自定义 sink 有这么几个疑问 > 1. 自带的 sink 得都改成 upsert 比如 jdbc/es > 2. 这样 append/upsert 代码有大量重复 > 3. 和 flink 对 append/upsert 流的定义有冲突，有额外 hack 的解释成本 > 4. 得有地方另外指定 update key > > 这么做感觉会挖坑 > > 有其他办法吗 > > lec ssmi <[email protected]>于2020年5月7日 周四20:42写道： > >> 使用自定义的Table Sink就可以了啊. >> >> Luan Cooper <[email protected]> 于2020年5月7日周四 下午8:39写道： >> >>> Hi >>> >>> 有这么个场景，需要将 MySQL 的对一张表的更改 同步到 ElasticSearch 中，两个地方表的 PRIMARY KEY 都是 >> ID，SQL >>> 如下 >>> >>> INSERT INTO sink_es // 将更改同步 upsert 到 ES >>> SELECT * >>> FROM binlog // mysql 表的 binlog >>> >>> 假设对于 MySQL 中 id = 1 的变更有 10 条，需要在 ES 上都更新 id = 1 的一条记录 >>> 但是上面的 SQL 是做不到的，只会一直 Insert >>> >>> 如果想 Upsert ES 的话，就得让 Source 表变成 Upsert Mode，但是现在好像不支持？ >>> >>> 社区的 FLIP-87 >>> >>> >> https://cwiki.apache.org/confluence/display/FLINK/FLIP+87%3A+Primary+key+constraints+in+Table+API >>> 可以解决这个问题吗？ >>> >>> 感谢 >>> >>
