问题
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数据流，这样下游再用同样的
聚合逻辑，效果也是一样的。
去重语法：
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
Original: https://www.cnblogs.com/bigdata1024/p/16303383.html
Author: chaplinthink
Title: Flink如何处理update数据
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/683961/
转载文章受原作者版权保护。转载请注明原作者出处！