实时计算Flink版：电商场景实战之营销红包

背景

某商家双十一活动期间的“回血红包”营销方案。用户在消费达到一定金额后，商家为了促进用户继续消费会返给用户一定金额的“回血红包”。实时计算能够为您实时的去监控用户的消费金额，筛选满足营销红包发放条件的用户。

解决方案

• SQL结构图
• 数据源表
  • 系统订单源表
    说明 为了聚焦核心逻辑，订单数据格式做了大量精简，只保留了与案例有关的属性。

    CREATE TABLE dwd_tb_trd_pay_ri(
            biz_order_id    VARCHAR, -- '订单ID' 
            auction_id      VARCHAR, -- '商品ID'
            auction_title   VARCHAR, -- '商品标题' 
            buyer_id        VARCHAR, -- '买家ID'
            buyer_nick      VARCHAR, -- '买家昵称'
            pay_time        VARCHAR, -- '支付时间'
            gmt_create      VARCHAR, -- '创建时间'
            gmt_modified    VARCHAR, -- '修改时间'
            biz_type        VARCHAR, -- '交易类型'
            pay_status      VARCHAR, -- '支付状态' 
            `attributes`    VARCHAR, -- '订单标记'
            from_group      VARCHAR, -- '订单来源'
            div_idx_actual_total_fee   DOUBLE  --'成交金额'        
    ) WITH (
        type='datahub',
        endPoint='http://dh-cn-hangzhou.aliyun-inc.com',
        project='yourProjectName',-- '您的project'
        topic='yourTopicName',--'您的topic'
        roleArn='yourRoleArn',-- '您的roleArn'
        batchReadSize='500'
    );               

  • 退货源表
    说明 为了聚焦核心逻辑，订单数据格式做了大量精简，只保留了与案例有关的属性。

    CREATE TABLE dwd_tb_trd_rfd_ri(
            biz_order_id    VARCHAR, -- '订单ID' 
            auction_id      VARCHAR, -- '商品ID'
            auction_title   VARCHAR, -- '商品标题' 
            buyer_id        VARCHAR, -- '买家ID'
            buyer_nick      VARCHAR, -- '买家昵称'
            pay_time        VARCHAR, -- '支付时间'
            gmt_create      VARCHAR, -- '创建时间'
            gmt_modified    VARCHAR, -- '修改时间'
            biz_type        VARCHAR, -- '交易类型'
            pay_status      VARCHAR, -- '支付状态' 
            `attributes`    VARCHAR, -- '订单标记'
            from_group      VARCHAR, -- '订单来源' 
            refund_fee      DOUBLE   -- '退款金额'     
    ) WITH (
        type='datahub',
        endPoint='http://dh-cn-hangzhou.aliyun-inc.com',
        project='yourProjectName', --'您的project'
        topic='yourTopicName', --'您的topic'
        roleArn='yourRoleArn', --'您的roleArn'
        batchReadSize='500'
    );                   

• 数据结果表

  关系型数据库RDS结果表

  CREATE TABLE tddl_output(
      gmt_create   VARCHAR, --'创建时间'
      gmt_modified VARCHAR, --'修改时间'
      buyer_id     BIGINT, --'买家ID'
      cumulate_amount BIGINT, --'金额'
      effect_time BIGINT, --'支付时间'
      primary key(buyer_id,effect_time)
  ) WITH ( 
          type= 'rds', 
          url = 'yourDatabaseURL',  --'您的数据库url' 
          tableName = 'yourTableName', --'您的表名 '
          userName = 'yourUserName',  --'您的用户名' 
          password = 'yourDatabasePassword' --'您的密码' 
      );      

• 业务逻辑

  把订单表和退货表做UNION ALL操作，获取到用户购买的所有商品的信息，统计用户的真实的消费金额和明细。

  CREATE VIEW  dwd_tb_trd_and_rfd_pay_ri
  AS
  SELECT
  *
  FROM(
      (SELECT 
          `a`.biz_order_id biz_order_id, 
          `a`.auction_id auction_id, 
          `a`.auction_title auction_title, 
          `a`.buyer_id buyer_id, 
          `a`.buyer_nick buyer_nick, 
          `a`.pay_time pay_time, 
          `a`.gmt_create gmt_create, 
          `a`.gmt_modified gmt_modified, 
          `a`.biz_type biz_type, 
          `a`.pay_status pay_status, 
          `a`.`attributes` `attributes`, 
          `a`.from_group,        
          `a`.div_idx_actual_total_fee div_idx_actual_total_fee
      FROM 
          dwd_tb_trd_pay_ri `a` 
      )
      UNION ALL
      (
      SELECT 
          `b`.biz_order_id biz_order_id, 
          `b`.auction_id auction_id, 
          `b`.auction_title auction_title, 
          `b`.buyer_id buyer_id, 
          `b`.buyer_nick buyer_nick, 
          `b`.pay_time pay_time, 
          `b`.gmt_create gmt_create, 
          `b`.gmt_modified gmt_modified, 
          `b`.biz_type biz_type, 
          `b`.pay_status pay_status, 
          `b`.`attributes` `attributes`, 
          `b`.from_group,     
          `b`.refund_fee div_idx_actual_total_fee --退款取负值
      FROM 
          dwd_tb_trd_rfd_ri `b` 
      )
  );               

• 去重操作

  在订单表和退货表中可能会存在大量重复的数据。例如商品ID、商品名称等。用MIN函数是为了只取第一次来的参数值，从而过滤掉其他的信息，然后再通过订单号和支付状态做分组取出需要的商品ID、商品名称等。

  
  CREATE VIEW filter_hxhb_dwd_tb_trd_and_rfd_pay_ri_distinct AS 
  SELECT 
     biz_order_id biz_order_id, 
     pay_status pay_status,
     MIN(auction_id) auction_id, 
     MIN(auction_title) auction_title, 
     MIN(buyer_id) buyer_id, 
     MIN(buyer_nick)  buyer_nick, 
     MIN(pay_time)  pay_time, 
     MIN(gmt_create)  gmt_create, 
     MIN(gmt_modified)  gmt_modified, 
     MIN(biz_type)  biz_type, 
     MIN(attributes)  attributes, 
     MIN(div_idx_actual_total_fee)  div_idx_actual_total_fee
  FROM 
      dwd_tb_trd_and_rfd_pay_ri
  GROUP BY biz_order_id ,pay_status;              

• 数据汇总所有数据做汇总处理

  CREATE VIEW ads_tb_trd_and_rfd_pay_ri AS
  SELECT 
     MIN(gmt_create) gmt_create,  --'创建时间'
     MAX(gmt_modified) gmt_modified,  --'修改时间'
     CAST (buyer_id AS BIGINT) buyer_id,  --'购买ID'
     CAST (date_format(pay_time , 'yyyy-MM-dd HH:mm:ss' , 'yyyyMMdd') AS BIGINT) as effect_time, --'购买时间'
     SUM(CAST(div_idx_actual_total_fee/100 AS BIGINT)) cumulate_amount  --'总的金额'
  FROM 
     filter_hxhb_dwd_tb_trd_and_rfd_pay_ri_distinct 
  GROUP BY 
     buyer_id,date_format(pay_time , 'yyyy-MM-dd HH:mm:ss' , 'yyyyMMdd');                 

  Q：为什么取MAX、MIN？

   MIN(gmt_create) gmt_create, --'创建时间'
   MAX(gmt_modified) gmt_modified, --'修改时间'                  

  A：MIN(gmt_create)是指的订单的第一笔订单的创建时间，MAX(gmt_modified)是指的最后一比订单的时间。根据订单的业务逻辑您可以用MAX和MIN来取相应的值。

  说明 如果时间字段不是BIGINT类型的可以用相应的内置函数做转换，详情请参见内置函数。
• 数据入库

  把统计好的数据插入RDS结果，再由其他的推送软件把相应的红包数发放给满足条件的用户。

  INSERT INTO tddl_output
  SELECT
      gmt_create,
      gmt_modified,
      buyer_id,
      cumulate_amount,
      effect_time
  from ads_tb_trd_and_rfd_pay_ri
  where cumulate_amount>0;               

  Q：为什么要取总的金额大于0的数？

  cumulate_amount>0      

  A：是为了过滤掉在上一步做UNION ALL的退货商品的金额。

总结

Q：在大量的订单和退货单中怎样才能清洗出我们所需要的数据？

A：在真实的购买记录中会存在大量的购买量和退货量，用UNION ALL把两张或者是多张表合并成一张大表，然后再通过具体的业务逻辑去重、聚合操作。最后把用户所有的真实订单交易金额写入存储中为后续的红包推送做准备。

DEMO示例以及源代码