（二）基于商品属性的相似商品推荐算法——Flink SQL实时计算实现商品的隐式评分

系列随笔：

提取码：4tds

注：如果你没有使用日志埋点和实时计算（接口直接累计也是可行的），你可以直接跳到这一节~

Flink SQL实时计算实现商品的隐匿评分

一、导入log service日志源表

二、导入评分配置维度表（用户行为的评分配置）

三、导入用户商品评分维表

四、用户评分结果表

四、预处理日志数据

-- 处理日志数据
CREATE VIEW probe_log0_view AS
SELECT
    t1.cid,
    CAST(memberCode as INT) as memberCode,
    t1.event,
    t1.eventApp,
    TO_TIMESTAMP(CAST(CAST(__timestamp__ as DOUBLE) as BIGINT)*1000) as eventTime,
    CAST(IF (SUBSTRING(t1.eventProps,0,1)='%', REGEXP_EXTRACT(t1.eventProps, concat(t2.code_name,'\\%22:(\\d+),'), 1), JSON_VALUE (t1.eventProps, concat('$.',t2.code_name))) as INT) as goodsCode,
    t2.score
FROM
    probe_log0 t1
    LEFT JOIN rc_config_dimension FOR SYSTEM_TIME AS OF PROCTIME() AS t2
    ON t1.event=t2.event AND t2.status=1
WHERE
    t1.event IN ('viewGoods','shareGoods','collectGoods','addToCart');

注：eventProps为埋点的扩展json数据，因为小程序的埋点不太规范，所以加了额外的判断；正常来说，直接使用 JSON_VALUE 函数即可

五、写入结果表

-- 入库
INSERT INTO rc_member_goods
    (member_code,
    cid,
    goods_code,
    score,
    update_time)
SELECT
    t1.memberCode,
    t1.cid,
    t1.goodsCode,
    CAST(IF(t2.score IS NOT NULL, t2.score, 0) + SUM(t1.score) as INT) AS score,
    MAX(t1.eventTime) as update_time
FROM
    probe_log0_view t1
    LEFT JOIN rc_member_goods_dimension FOR SYSTEM_TIME AS OF PROCTIME() AS t2
    ON t1.memberCode=t2.member_code AND t1.cid=t2.cid AND t1.goodsCode=t2.goods_code
WHERE
    t1.goodsCode IS NOT NULL
    AND (t1.eventTime > t2.update_time OR t2.update_time IS NULL)
GROUP BY
    t1.memberCode,
    t1.cid,
    t1.goodsCode,
    t2.score;