一、背景

? ? ? ?在电商平台中，用户通过利用多个账号领取优惠券，薅平台补贴力度比较大的商品，为了躲避平台的同一个地址的规则：平台会限制一个收件地址在同一天内只能下一单的规则，会加一些干扰项来逃避平台的规则，比如：上海市浦东新区张江路88号、上海市浦东新区张江路88号哈哈哈哈，其实这两个地址其实代表的同一个地址，因此本文讲述一种方法，来实时检测出这部分订单并加以拦截。

? ? ? ?本文基于公司的实时计算平台，借助Flink实时计算能力，实现对实时地址相似度的计算。

二、实现流程

? ? ? 以上流程图我们公司实时计算平台的应用体系。最终的应用平台可以提供实时数据展示、写入redis中供下游服务进行调用、输出风控系统提供告警。目前的应用是这样的，后续还规划给用户实时打上标签，提供数据到标签系统。

三、业务逻辑计算

3.1 业务逻辑

? ? ? ? 针对特别的商品，计算一个区域内，下单地址与区域内其他地址之间的相似度，该值达到一定的阈值，就对该数值进行累加，当累加的数量超过一定数量后，就进行拦截退款操作，当然实时计算平台只负责进行计算，具体拦截操作需将结果传给风控系统。

3.2 地址相似度计算

该算法是用python开发的UDF函数，采用jieba分词先对地址进行分词，然后在计算地址的相似度，具体实现方法可以参见本人的另一篇博客：

https://blog.csdn.net/weixin_37536446/article/details/81284025?spm=1001.2014.3001.5501

3.3 数据存储

? ? ? 由于最终的结果是要给到风控系统进行调用，我们选择存储的介质是Redis，选择redis的原因是：随着订单源源不断的进入到实时平台，因此对于每一单来说，与其相似地址的个数是时时不断变化的，而Redis的k-value结构，会实时更新计算的结果。

四、使用Flink-sql中所遇到的坑

1、flink-sql中的部分语法与hive-sql存在区别

对时间属性字段以外的字段进行GROUP BY （滚动窗口、滑动窗口或会话窗口中的GROUP BY 除外）操作。
双流join操作
复杂事件处理（CEP）语句中的MATCH_RECOGNIZE操作
OVER窗口中的PARTITION BY?操作。
UNION ALL操作。UNION = RETRACT + UNION ALL

如果经过以上操作后，继续使用时间属性字段进行窗口函数运算，会出现类似

Org.apache.flink.table.api.ValidationException:Window can only be defined over a time attribute column.的报错。

2、Flink实现3个实时流join,left join,right join

整体思路：

设置相同的时间类型
设置相同的时间窗口，这样就会到达相同窗口，3个实时流会同时触发

由于Flink不支持3个实时流同时join，需要先把两个实时流join完成的结果，再跟第三个实时流join。

3、时间函数

不要使用CURRENT_DATE,CURRENT_TIMESTAMP等时间函数，这些函数默认的值都是UTC+0的值，在统计上有时差（8小时）

原函数	替换函数
CURRENT_DATE	DATE_FORMAT(LOCALTIMESTAMP,'yyyy-MM-dd')
CURRENT_TIMESTAMP	LOCALTIMESTAMP

4、计算两个时间戳相差的秒数

SELECT logtimeTs
      ,unix_timestamp(logtimeTs) AS logtimeTs_timestamp  -- 转化后是毫秒单位
      ,pageTs
      ,unix_timestamp(pageTs)    AS pageTs_timestamp     -- 转化后是毫秒单位
  FROM dw_act.ods_ubtspm_h5

注：两个时间戳相减后，得出相差的毫秒数

5、Flink实现按照一段窗口进行数据统计

-- 统计一分钟内，消耗的现金、卡券等指标
SELECT trade_id
      ,item_sku_id
      ,TUMBLE_START(gmt_modified,INTERVAL '1' MINUTE) 
      ,TUMBLE_END(gmt_modified,INTERVAL '1' MINUTE)
      ,SUM(jiankangjin_mid/100)           AS health_gold
      ,SUM(cash)                          AS cash
      ,SUM(card_fee_mid)                  AS card_fee
      ,MAX(rate_merchant_jiankangjin_mid) AS rate_merchant_jiankangjin  
  FROM
  (
      SELECT trade_id
            ,item_sku_id
            ,gmt_modified
            ,proctime()    AS proctime
            ,coalesce(cast(get_json_object(ext_props,'$.ownedRatio') AS float),0) AS ownedRatio
            ,jiankangjin_mid
            ,cash_mid
            ,rate_merchant_jiankangjin_mid
            ,card_fee_mid
        FROM ods_tddl_tcorder_tc_order_payment_mod    
  ) a 
  GROUP BY trade_id,item_sku_id

? ? ? ?实时计算，在现实业务中的应用越来越强大，如实时计算广告投放广告效果的监控，及时调整渠道的广告投放、实时监控风控系统的拦截情况、防止被薅羊毛。

后续更新脚本及资源调优篇。