在Apache Flink中手动更新状态的最佳方法是什么？
我在股票市场项目中使用Apache Flink来计算当前的价格变化。公式是
price_change = (current_price - previous_close_price) / previous_close_price
previous_close_price是交易所前一天的证券收盘价。在市场开放前的每一天，我都需要更新previous_close_price。
现在我想出了几个解决方案，但我不知道哪个是最好的。
- 存储previous_close_price在redis中并在每次计算中获取价格。更新价格既简单又灵活，但此解决方案可能会降低性能。
- 将状态的TTL设置为1天。旧状态到期时获取新状态。但它不灵活，因为TTL是硬编码的。
- 广播状态模式。我不确定这个解决方案是否有效。
- 发送特殊消息给flink。当flink收到消息时，它会更新previous_close_price。
展开
收起
1
条回答
写回答
写回答
-
flink小助手会定期更新直播回顾等资料和文章干货，还整合了大家在钉群提出的有关flink的问题及回答。"我建议在＃4上使用一个变体：
有两个来源，一个仅用于收盘价，另一个用于交易流。通过安全性键入两个流，并将它们与CoProcessFunction连接。将previous_close_price存储在CoProcessFunction中的键控状态。
每天，在市场开放之前，以最新的收盘价流动。
这可以使用RichCoFlatMap完成，但我建议使用CoProcessFunction，因为您可能希望使用侧输出来报告错误（例如缺少previous_close_price的证券）。
至于其他方法：- 我认为将previous_close_price数据保存在外部数据存储中没有任何好处。
- 我觉得这不太好用。没有可用于触发新数据加载的钩子，而且只有在访问状态时才会清除状态。
- 这对广播状态来说不是一个好的用例，除非集群中的每个人都需要知道所有证券的收盘价。"
2019-07-17 23:16:45赞同 展开评论 打赏
问答分类：
版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
相关问答
