问题
Flink实时统计GMV,如果订单金额下午变了该怎么处理
具体描述
- 实时统计每天的GMV,但是订单金额是会修改的。
- 订单存储在mysql,通过binlog解析工具实时同步到kafka.然后从kafka实时统计当日订单总额。
- 假设订单009 上午10点生成，金额为1000. 生成一条json数据到kafka ,GMV实时统计为1000。
- 然后下午15点，009订单金额被修改为500。数据生成json也会进入kafka. 这时如果不减去上午已经统计的金额。那么总金额就是错的。
根据 update /delete 要写这个减去的逻辑。
按日去重是不行了，因为是增量处理, 上午的数据已经被处理了不能再获取了。
解决思路
- 首先版本是1.11 ， 可以直接用binlog format，这样数据的修改其实会自动对应到update_before和update_after的数据，这样Flink 内部的算子都可以处理好这种数据，包括聚合算子。比如你是select sum(xxx) from T group by yyy这种，那这个sum指标会自动做好这件事。
- 如果不用binlog模式，只是取最新的数据来做聚合计算，也可以用去重算子[1] 将append数据流转成retract数据流，这样下游再用同样的 聚合逻辑，效果也是一样的。
去重语法：
代码语言：javascript复制SELECT [column_list]
FROM (
SELECT [column_list],
ROW_NUMBER() OVER ([PARTITION BY col1[, col2...]]
ORDER BY time_attr [asc|desc]) AS rownum
FROM table_name)
WHERE rownum = 1- ROW_NUMBER(): 每一行分配一个唯一的，序列数字，从1开始
- PARTITION BY col1[, col2...]: 指定分区列 i.e. 去重key.
- ORDER BY time_attr [asc|desc]: 指定排序字段, 必须是一个时间属性. Currently Flink 支持 processing time 和 event time 属性. Ordering by ASC 意为保留第一行, ordering by DESC 意为 保留最后一行.
- WHERE rownum = 1: The rownum = 1 是必须的，对于Flink识别这个是去重的查询语句
只要source端产生了changelog数据，后面的算子是可以自动处理update消息的，简单理解，你可以认为:
- append / update_after 消息会累加到聚合指标上
- delete / update_before 消息会从聚合指标上进行retract
Reference
- https://nightlies.apache.org/flink/flink-docs-master/zh/docs/dev/table/sql/queries/deduplication/
- https://developer.aliyun.com/article/782653
