Hi, 问题一:删除数据可不单单只是retract stream的功能。upsert stream是当下游具有按key覆盖的功能时的特殊优化,除了按key覆盖外,它也需要在上游retract时删除数据,意思是upsert stream也有retract的input数据的。JDBC实现的是upsert stream的消费。
问题二:正确数据应该是: 1 {"order_id":1,"tms_company":"zhongtong"} 数据库1条记录: zhongtong 1 2 {"order_id":1,"tms_company":"yuantong"} 数据库1条记录: yuantong 1 ( 删除 zhongtong 1) 3 {"order_id":2,"tms_company":"yuantong"} 数据库1条记录: yuantong 2 ( 删除yuantong 1) 4 {"order_id":2,"tms_company":"zhongtong"} 数据库2条记录: yuantong 1, zhongtong 1 ( 删除yuantong 2) 你用了什么dialect?是不是mysql? Flink JDBC的Mysql用了DUPLICATE KEY UPDATE的语法来更新数据。 看起来这个语法在RDS没有建主键或者唯一键时可能不会去覆盖老数据?尝试创建下主键或唯一建? Best, Jingsong Lee On Wed, May 6, 2020 at 10:36 AM wangl...@geekplus.com.cn < wangl...@geekplus.com.cn> wrote: > > 更新键是 tms_company, 但这是通过双层的 group 实现了回退功能,总共就 两个 order_id, order_id 对应的 > tms_company 是有变化的。 > 我不是很明白这种回退的具体原理,为什么有的会删除,有的就没有删除。 > > > > > wangl...@geekplus.com.cn > > 发件人: Michael Ran > 发送时间: 2020-04-30 17:23 > 收件人: user-zh > 主题: Re:FlinkSQL Retraction 问题原理咨询 > > > > 指定的更新键是tms_company? > > > 结果是: > yuantong:2 > zhongtong:2 > > > > > > > > > > > > > 在 2020-04-30 17:08:22,"wangl...@geekplus.com.cn" <wangl...@geekplus.com.cn> > 写道: > > > >自己实现了一下 https://yq.aliyun.com/articles/457392/ 菜鸟物流订单统计的例子,读 kafka 写到 > RDS, RDS 表没有主键,也没有唯一键。 > > > >INSERT INTO table_out select tms_company, count(distinct order_id) as > order_cnt from > > (select order_id, LAST_VALUE(tms_company) AS tms_company from > dwd_table group by order_id) > > group by tms_company; > > > > > >总共发送了 4 条消息,顺序如下: > > > >1 {"order_id":1,"tms_company":"zhongtong"} 数据库1条记录: zhongtong 1 > > > >2 {"order_id":1,"tms_company":"yuantong"} 数据库1条记录: yuantong 1 > (上一条记录被删除了) > > > >3 {"order_id":2,"tms_company":"yuantong"} 数据库2条记录: yuantong 1, > yuantong 2 (增加了条记录,没有删除) > > > >4 {"order_id":2,"tms_company":"zhongtong"} 数据库4条记录: yuantong 1, > yuantong 2, yuantong 1, zhongtong 1 (增加了两条记录,没有删除) > > > > > >问题一: > > 第 2 条消息发送后,数据库的上一条记录被删除了。我的理解这应该是 RetractStream 的功能。当我看源码 > https://github.com/apache/flink/tree/master/flink-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc > 这里也没有 RetractionStream 的实现,哪里的代码把他删除了呢? > > > >问题二: > > 第 3 条记录来了后,直接在数据库增加了 一条 yuantong 2, 为什么没把 yuantong 1, 删除呢? > > 第 4 条记录来了后,又在数据库里加了两条记录,为什么也没有删除之前的呢? > > > >谢谢, > >王磊 > > > > > > > >wangl...@geekplus.com.cn > -- Best, Jingsong Lee