实时计算如何帮助淘宝实现在线「实时选品」？

一、背景介绍

二、解决方案

2.1 问题分析

• 实时触发问题：流计算必须要有触发数据源，用户提交的特征数据存储在idb中，idb如何与Blink计算流程建立关联？
• 中间状态存储问题：Blink计算过程中，依据业务场景需要记录上一次中间计算状态，如何存储这些中间状态，并在需要时实时读取？
• 实时生效问题：Blink计算结果需要最终在搜索引擎生效，Blink如何跟搜索引擎交互，使计算结果实时生效？
• 增量问题：如果没有增量过程，离线全量切换数据期间，会覆盖部分已更新的数据，入如何增量追数据？

• TT日志：阿里日志采集系统，用户可以对日志进行订阅，TT与idb(数据库)、Blink有较好支持，是idb和Blink交互的重要媒介。
• Hbase：开源的非关系型分布式数据库，与Blink有较好的接口交互，可用于存储和读取中间计算状态。
• Swift：阿里搜索事业部自主研发的消息系统，目前主搜索实时都是基于该系统进行消息传输的，可基于该系统解决引擎实时生效和增量问题。

2.2 实现流程

• 用户提交选品特征数据，数据保存到idb(数据库)并同步到TT日志。
• TT日志更新会触发Blink任务，日志解析节点会先解析TT日志，获取选品特征数据。
• Query拆分节点先预估SPU数量，然后根据SPU数量确定并发请求数，并拼接Sp参数。
• 请求Sp服务节点并发请求Sp服务，获取SPU信息。
• 数据计算节点从Hbase中读取中间状态数据，根据业务逻辑进行计算。
• 计算结果回写Hbase数据库，用于下次计算使用。
• 回写TT节点和Swift消息节点同时回写TT和Swift。
• dump接受Swift消息，将数据更新至引擎，实现数据实时生效。
• TT记录计算结果，回写ODPS，用于离线全量计算。

2.3 实现细节

• 节点分析：“请求Sp节点”的业务场景是一个“一对多”的过程，因此采用UDTF类型实现。
• 封装UDTF类：该类需要继承TableFunction,其中T为自己定义的pojo，用于向下个运行节点传递。
• 节点输出：需要定义自己的Pojo类(上一步提到的T)，这样节点的输出才能在下个节点中看到。
• 主函数流程串联：Blink开发流程需要一个主函数将各个计算节点关联起来，达到流计算的目的，建议主函数用Scala语言开发，代码更加通俗易懂。

2.4 参考代码

public class SearchEngineUdtf extends TableFunction<EngineFields> {

    private static final Logger logger = LoggerFactory.getLogger(SearchEngineUdtf.class);

    /**
     * 请求引擎获取召回字段
     * @param params
     */
    public void eval(String params) {
        SpuSearchResult<String> spuSearchResult = SpuSearchEngineUtil.getFromSpuSearch(params);
        if(spuSearchResult.getSuccess()){
            //结果解析
            JSONObject kxuanObj = SpuSearchEngineUtil.getSpResponseJson(spuSearchResult, "sp_kxuan");
            if(null == kxuanObj || kxuanObj.isEmpty()){
                logger.error("sp query: " + spuSearchResult.getSearchURL());
                logger.error(String.format("[%s],%s", Constant.ERR_PAR_SP_RESULT,"get key:sp_kxuan data failed! "));
            }else {
                List<EngineFields> engineFieldsList = SpuSearchEngineUtil.getSpAuction(kxuanObj);
                //并发输出到数据流
                for(EngineFields engineFields : engineFieldsList){
                    collect(engineFields);
                }
            }
        }else {
            logger.error(String.format("[%s],%s",Constant.ERR_REQ_SP, "request SpuEngine failed!"));
        }

    }
}

2.5 运维发布

三、成果总结

四、作者简介

如果您有实时报表／实时数据大屏／实时金融风控／实时电商推荐等相关实时化数据处理需求，可以加入如下钉钉交流群！