FLINK SQL实战案例之商品销量实时统计

问题导读

1.本文的业务包含哪些流程？
2.本文难点在什么地方？
3.如何通过flink sql实现商品销量实时统计？
 

1、案例背景介绍
互联网电商往往需要对订单商品销量实时统计，用于实时大屏展示，库存销量监控等等。本文主要介绍如何通过flink sql的方式进行商品实时销量的统计。

业务流程介绍：

1.使用otter采集业务库binlog数据输出到kafka

2.flink读取kafka数据进行商品销量统计

3.统计结果输出到mysql

4.下游业务系统直接读取mysql数据

业务需求介绍：

根据订单创建时间统计商品每天的实时销量，不包含取消订单的商品

2、准备工作
将mysql订单相关的binlog日志实时同步到kafka对应的Topic，然后创建对应的flink table source表。

为了简化需求，下面的订单表和订单明细表只列出主要的字段。

订单主表：orders
 

订单明细表：order_detail
 

3、难点解析
同一个订单会有多次业务操作（例如下单、付款、发货,取消等等），每一次业务操作都会导致订单状态发生变化，并且每次变化订单表对应的Binlog日志会产生一条订单号相同的数据。如果我们不做处理直接关联聚合查询的话会导致数据重复统计结果不正确。因此我们需要了解业务系统都有哪些操作会对订单主表和订单明细进行更新操作。

假设业务系统数据变更是这样的：

用户下单后新增订单主表和订单明细表数据
后续的业务操作只会更新订单主表数据，订单明细表数据不会更新变化
数据每次更新update_time字段都会同时变化
再来看一下我们的需求如何处理：

需求的要求是当订单创建后我们就要统计该订单对应的商品数量，而当订单状态变为取消时我们要减掉该订单对应的商品数量。根据需求和订单数据更新的特点，这里需要用到flink回撤流的特性来处理该需求。flinksql可以使用row_number() over(partition by order_no order by update_time desc) 通过限制 where rn=1来获取同一订单的最新状态数据，然后和订单明细表进行关联求和。flinksql会自动更新统计结果。

4、编写业务逻辑

--订单主表source table
CREATE TABLE orders 
            (
               order_no     string,
               order_state  int,
               pay_time     string,
               create_time  string,
               update_time  string
             ) 
       WITH (
               'connector.type' = 'kafka',       
               'connector.version' = 'universal', --kafka版本    
               'connector.topic' = '_tporders',--kafkatopic
               'connector.properties.zookeeper.connect' = 'localhost:2181', 
               'connector.properties.bootstrap.servers' = 'localhost:9092',
               'connector.properties.group.id' = 'testGroup',
               'connector.startup-mode' = 'latest-offset',
               'format.type' = 'json' --数据为json格式             
             )
             
--订单明细表source table
CREATE TABLE order_detail 
            (
               order_no     string,
               product_code string,
               quantity     int,
               create_time  string,
               update_time  string
             ) 
       WITH (
               'connector.type' = 'kafka',       
               'connector.version' = 'universal', --kafka版本    
               'connector.topic' = 'tp_order_detail',--kafkatopic
               'connector.properties.zookeeper.connect' = 'localhost:2181', 
               'connector.properties.bootstrap.servers' = 'localhost:9092',
               'connector.properties.group.id' = 'testGroup',
               'connector.startup-mode' = 'latest-offset',
               'format.type' = 'json' --数据为json格式             
             )
--mysql统计结果表sink table
--mysql建表时指定主键为order_date,product_code,flink写入数据时相同主键会进行更新
CREATE TABLE product_sale
             (
              order_date string,
              product_code string,
              cnt int
              ) 
         WITH (
           'connector.type' = 'jdbc', 
           'connector.url' = 'jdbc:mysql://localhost:3306/flink?serverTimezone=UTC&useSSL=true', 
           'connector.table' = 'order_state_cnt', 
           'connector.driver' = 'com.mysql.cj.jdbc.Driver', 
           'connector.username' = 'root',
           'connector.password' = 'root',
           'connector.write.flush.max-rows' = '1',--默认每5000条数据写入一次，测试调小一点
           'connector.write.flush.interval' = '2s',--写入时间间隔
           'connector.write.max-retries' = '3'
         )
         
 --统计商品销量并写入mysql        
insert into product_sale 
select create_date,product_code,sum(quantity)
from (select t1.order_no,
             t1.create_date,
             t2.product_code,
             t2.quantity
       from (select order_id,
                    order_status,
                    substring(create_time,1,10) create_date,
                    update_time ,
                    row_number() over(partition by order_no order by update_time desc) as rn
              from orders
              )t1
       left join order_detail t2
            on t1.order_no=t2.order_no
      where t1.rn=1--取最新的订单状态数据
      and t1.order_status<>0--不包含取消订单
   )t3
 group by create_date,product_code

5数据测试

假设在13点创建了两个订单，数据如下：

订单主表数据：

订单明细数据：

统计结果：

然后订单号为order1的订单取消了：

统计结果：