flink-connector-redis自定义连接器,在实现DynamicTableSink接口时, 通过重写getChangelogMode方法,设置ChangelogMode.upsert()模式,以更新的方式进行数据写入, 使用的窗口时cumulate函数 开1天的窗口,小窗口1分钟计算一次
问题: 测试发现,每1分钟都会输出一次,落地的数据一样, 根据upsert()模式,正常来说一样的数据应该不会触发redis存储,这个是什么情况?*来自志愿者整理的flink邮件归档
可以在接受 sink 数据的方法里 System.out.println 看看是否真的每分钟收到一次数据。
如果是的话,应该是设置了 state ttl 的缘故。为了防止数据过期,group agg 就算每次收到同样的 key
value,仍然会往下游发数据(详见 GroupAggFunction#processElement),所以这是一个预期行为。*来自志愿者整理的FLINK邮件归档
版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。