实时数仓建设第2问：怎样使用flink sql快速无脑统计当天下单各流程（已发货，确认收货等等）状态的订单数量

实时统计当天下单各流程状态(已支付待卖家发货，卖家通知物流揽收，待买家收货等等)中的订单数量。
订单表的binlog数据发送到kafka,flink从kafka接受消息进行指标统计。因为每笔订单的状态会发生变化，比如上午为【已支付待卖家发货】，这个时候【已支付待卖家发货】指标数要+1，下午订单的状态变更为【卖家通知物流揽收】，这个时候【卖家通知物流揽收】指标数要+1，而【已支付待卖家发货】指标数要-1。

如果采用Java代码编程，那么需要深入理解业务考虑每种状态变更，编写大量if逻辑稍有遗留就会统计错误结果。但是如果使用flink sql是不是就不需要考虑这些业务问题了？
要想使得

select order_status,count(order_no) from order group by order_status

能够得到我们期望的结果，必须使得进入该SQL的数据流由append流变成update/retract流。否则就要根据binlog消息的update /delete情况编写这个减去的逻辑。

解决思路

• flink数据消费不走kafka，直接使用flink cdc去消费数据库的binlog日志。
• flink数据消费走kafka,那么kafka ddl使用canla-json format。
• 如果当前flink版本不支持canla-json format，那么就需要将soucre出来的append流转换为update/retract流后进入我们的聚合SQL算子中。

只要source端产生了changelog数据，后面的算子是可以自动处理update消息的，你可以认为:

• insert/ update_after 消息会累加到聚合指标上
• delete / update_before 消息会从聚合指标上进行retract

专栏初衷:

• 要想快速建设实时数仓，对齐离线数仓分层，首选Flink SQL，相比datastream 代码，Flink SQL可大幅提升10倍实时数仓建设落地时间。
• 作者位于大厂实时数仓团队，目前运行实时任务3000+，实时集群规模2万CU，集群checkpoint峰值5TB,单任务最大QPS峰值50W。
• 本专栏将分享作者在实时数仓建设过程中遇到的细节点，帮组大家快速建设实时数仓。

作者信息:

• 1.《深入理解Flink核心设计与实践原理》一书作者
• 2.GitHub 热门项目 fink-boot（800+） 开发者，致力于flink与spring生态集成