一种基于分布式实时计算的能源商品及能源需求的精准推送方法

摘要

一种基于分布式实时计算的能源商品及能源需求的精准推送方法，包括以下步骤：步骤一：用户操作日志采集，负责采集用户在网页端或APP端产生的各类点击日志；步骤二：数据缓冲，与日志采集部分对接，利用消息中间件接收并缓存日志消息，与实时计算部分对接，将日志消息传递给计算程序；步骤三：混合分布式计算，综合利用各种数据，实时产生针对能源用户或能源服务商的推送结果；步骤四：数据存储，用于存储能源用户信息等，便于步骤一种的用户调取实用。该推送方法实现针对能源商品及能源需求的智能化、个性化、精准化的推送系统。帮助能源用户快速找到自己需要的能源商品，帮助能源服务商快速找到目标能源客户，促进供需共赢。

说明书

技术领域

本发明涉及大数据推送技术领域，具体为一种基于分布式实时计算的能源商品及能源需求的精准推送方法。

背景技术

随着能源行业的飞速发展,电子商务在能源市场的参与度也越来越高.打造一个能源市场的“淘宝”，让其中用户发布需求，服务商发布商品，则可以充分发挥市场服务对接的纽带作用，会吸引全社会能源用户和各类能源服务商在此开展服务。而市场的商品推送系统则是其中的关键,推送系统是自动联系用户和商品的一种工具，它通过收集市场与用户的各类信息，然后根据这些行为数据，分析出一定的规则或者直接对用户对其他物品的喜好进行预测计算。从而主动给用户推送可满足他们兴趣和需求的信息。每个用户所得到的推送信息都是与自己的行为特征和兴趣有关的，而不是笼统的大众化信息。一是可以反映用户对物品的真实喜好,对用户线上和线下行为进行深度洞察,提高用户时间效率.二是提高了商家的个性化服务，进一步提升顾客在店铺的消费体验。

现有推送系统的推送结果产生基于全品类商品热度统计，没有采用分析用户的历史行为来对用户的兴趣进行分析，无法针对具体用户产生个性化、精准化的推送结果。而且一般的实现方案是基于离线统计实现，推送结果产生存在滞后性。推送结果类型单一，没有针对用户当下及历史感兴趣商品做区别推送。其次现有推送系统没有综合考虑商品评分、用户消费水平等重要影响因素。再其次没有针对能源用户需求的推送实现方案。

所以为了让商家更了解用户，推送系统既要服务于电商，更要服务于用户，提高推送结果精准度是现有技术中亟待解决的技术问。

通过公开专利检索，发现以下对比文件：

CN111061807A-公开了一种分布式数据采集分析系统及方法、服务器及介质，其通过不同的采集方式全面对数据进行采集，采用分布式架构有效的横向扩展数据量的增加，且kafka数据库高吞吐量优势满足直接数据采集和传输需求。数据提取单元采用脚本语言实现，能够在线及时修改调试，同时规则引擎子单元的使用能够动态分配数据,灵活性的数据分发为系统的扩展性提供了基础。

经分析，上述专利中的分布式数据采集分析系统与本申请相比，在实时计算功能等方面存在较大差异，因此不影响本申请的信用行。

发明内容

本发明的目的在于克服现有技术的不足之处，提供一种基于分布式实时计算的能源商品及能源需求的精准推送方法，该方法实现针对能源商品及能源需求的智能化、个性化、精准化的推送系统。帮助能源用户快速找到自己需要的能源商品，帮助能源服务商快速找到目标能源客户，促进供需双方合作，实现双方共赢。

一种基于分布式实时计算的能源商品及能源需求的精准推送方法，包括以下步骤：

步骤一：用户操作日志采集，负责采集用户在网页端或APP端产生的各类点击日志；

步骤二：数据缓冲，与日志采集部分对接，利用消息中间件接收并缓存日志消息，与实时计算部分对接，将日志消息传递给计算程序；

步骤三：混合分布式计算，将现有独立式的流计算与批计算融合为混合分布式计算，处理实时流数据的同时利用自定义累加器在内存中计算并维护批量数据统计指标，实时产生针对能源用户或能源服务商的推荐结果；

步骤四：数据存储，用于存储能源用户信息、能源服务商信息、商品信息、合同信息、日志信息、推荐结果信息，便于步骤一种的用户调取实用。

而且，步骤一中日志采集部分主要收集用户在系统网页及APP页面上的点击日志，包括能源用户点击能源商品的日志、能源服务商点击能源需求的日志、能源用户搜索能源商品的日志、能源服务商搜索能源需求的日志、能源用户生成能源商品合同的日志、能源服务商生成能源需求合同的日志、能源用户对能源商品评分的日志、能源用户对能源服务商评分的日志及能源用户发布能源需求的日志。

而且，步骤二中的数据缓冲部分利用Kafka消息中间件接收日志采集部分所产生的日志。

而且，步骤三中的流批一体化分布式计算同时使用Spark流处理及批处理算子，流处理算子负责拉取消息中间件中的实时数据并计算实时数据指标，批处理算子负责计算并维护批量数据统计指标，启动时批处理算子先从数据库中读取计算所需各种数据，主要包括：

(1)用户历史操作日志，包括能源用户点击能源商品的日志、能源服务商点击能源需求的日志、能源用户生成能源商品合同的日志、能源服务商生成能源需求合同的日志；

(2)能源商品信息，包括能源商品ID、能源商品综合评分、能源商品价格；

(3)能源用户标签信息、标签对应的商品类型、能源用户名称。

而且，步骤四获取步骤三中数据后对其进行处理，并将处理结果保存在内存中，主要包括：

(1)基于历史点击或搜索日志，按照能源商品或能源需求、操作日志类型、用户名称、能源商品或能源需求类型进行聚合，统计能源用户针对能源商品的点击次数和搜索次数及能源服务商针对能源需求的点击次数和搜索次数；

(2)基于历史成交合同，按照能源商品或能源需求、能源用户或能源服务商ID、能源商品或能源需求类型，计算能源用户针对能源商品类型的平均消费金额、平均评分及能源服务商针对能源需求类型的平均得分。

而且，步骤四的历史数据处理完毕后，开始拉取Kafka中的实时日志消息，每隔一段时间间隔拉取一批消息并产生推荐结果，该时间间隔根据实际情况进行配置；根据日志中消息类型标记的不同进行不同的处理：

(1)点击或搜索日志类消息，按照能源用户或能源服务商、能源商品类型或能源需求类型统计在该批次消息中能源用户点击和搜索各类能源商品的次数及能源服务商点击和搜索各类能源需求的次数；根据统计结果计算能源用户当前最感兴趣的能源商品类型，再根据能源用户对该能源商品类型的评分、能源用户平均消费金额及各个能源商品综合评分、价格，计算出能源用户当前可能最感兴趣的能源商品；同时将本批次统计的次数累加到历史总次数中，基于历史总次数再计算出能源用户历史最感兴趣的能源商品，基于用户标签计算出能源用户可能感兴趣的能源商品；对于能源服务商点击和搜索各类能源需求的次数只累加到历史总次数中，用于能源需求推荐；

(2)合同类消息，按照能源用户或能源服务商、能源商品类型或能源需求类型统计该批次消息中能源用户对各类能源商品生成合同的金额及能源服务商对各类能源需求生成合同的次数，用于计算能源用户对各类能源品的平均消费金额、评分和能源服务商在各类能源需求的平均得分；

(3)能源需求类消息，根据能源服务商对各类需求点击和搜索次数，计算出对该类能源需求最感兴趣的能源服务商，再根据能源服务商在该类能源需求的得分筛选出推荐结果。

而且，步骤四的数据存储部分采用Oracle关系型数据库，主要存储内容包括：能源用户信息、能源服务商信息、商品信息、合同信息、日志信息及推荐结果信息。

本发明的优点和技术效果是：

本发明的一种基于分布式实时计算的能源商品及能源需求的精准推送方法，其优势在于：

(1)本方法采用分布式实时计算技术实现，基于用户实时操作日志，实时产生推送结果，具有较高的实时性。

(2)本方法基于能源用户产生的点击日志、合同日志等，针对用户自身兴趣产生个性化、精准化的推送结果。

(3)本方法推送的结果结果分多种类型，有基于能源用户实时日志产生的能源用户当前最感兴趣推送结果，有基于能源用户历史日志、合同产生的能源用户历史最感兴趣推送结果，有基于能源用户标签产生的标签推送结果。

(4)推送结果综合考虑了商品综合评分、商品价格、商品上架时间、用户对商品的个人评分、用户的消费水平等因素，推送结果更精准，更符合用户需求。

(5)能够将能源用户发出的能源需求，精准推送给能源服务商，撮合双方合作。

本发明的一种基于分布式实时计算的能源商品及能源需求的精准推送方法，利用大数据分布式实时计算技术，根据能源用户操作日志、成交合同、用户标签等数据，实时计算出能源用户可能感兴趣的能源商品以及能源服务商可能满足的能源用户需求，并将能源商品推送给能源用户，将能源用户需求推送给能源服务商。

附图说明

图1为本发明数据流的结构示意图；

图2为本发明混合分布式计算示意图。

具体实施方式

为能进一步了解本发明的内容、特点及功效，兹例举以下实施例，并配合附图详细说明如下。需要说明的是，本实施例是描述性的，不是限定性的，不能由此限定本发明的保护范围。

一种基于分布式实时计算的能源商品及能源需求的精准推送方法，包括以下步骤：

步骤一：用户操作日志采集，负责采集用户在网页端或APP端产生的各类点击日志；

步骤二：数据缓冲，与日志采集部分对接，利用消息中间件接收并缓存日志消息，与实时计算部分对接，将日志消息传递给计算程序；

步骤三：混合分布式计算，将现有独立式的流计算与批计算融合为混合分布式计算，处理实时流数据的同时利用自定义累加器在内存中计算并维护批量数据统计指标，实时产生针对能源用户或能源服务商的推荐结果；

步骤四：数据存储，用于存储能源用户信息、能源服务商信息、商品信息、合同信息、日志信息、推荐结果信息，便于步骤一种的用户调取实用。

而且，步骤一中日志采集部分主要收集用户在系统网页及APP页面上的点击日志，包括能源用户点击能源商品的日志、能源服务商点击能源需求的日志、能源用户搜索能源商品的日志、能源服务商搜索能源需求的日志、能源用户生成能源商品合同的日志、能源服务商生成能源需求合同的日志、能源用户对能源商品评分的日志、能源用户对能源服务商评分的日志及能源用户发布能源需求的日志。

而且，步骤二中的数据缓冲部分利用Kafka消息中间件接收日志采集部分所产生的日志。

而且，步骤三中的流批一体化分布式计算同时使用Spark流处理及批处理算子，流处理算子负责拉取消息中间件中的实时数据并计算实时数据指标，批处理算子负责计算并维护批量数据统计指标，启动时批处理算子先从数据库中读取计算所需各种数据，主要包括：

(1)用户历史操作日志，包括能源用户点击能源商品的日志、能源服务商点击能源需求的日志、能源用户生成能源商品合同的日志、能源服务商生成能源需求合同的日志；

(2)能源商品信息，包括能源商品ID、能源商品综合评分、能源商品价格；

(3)能源用户标签信息、标签对应的商品类型、能源用户名称。

而且，步骤四获取步骤三中数据后对其进行处理，并将处理结果保存在内存中，主要包括：

(1)基于历史点击或搜索日志，按照能源商品或能源需求、操作日志类型、用户名称、能源商品或能源需求类型进行聚合，统计能源用户针对能源商品的点击次数和搜索次数及能源服务商针对能源需求的点击次数和搜索次数；

(2)基于历史成交合同，按照能源商品或能源需求、能源用户或能源服务商ID、能源商品或能源需求类型，计算能源用户针对能源商品类型的平均消费金额、平均评分及能源服务商针对能源需求类型的平均得分。

而且，步骤四的历史数据处理完毕后，开始拉取Kafka中的实时日志消息，每隔一段时间间隔拉取一批消息并产生推荐结果，该时间间隔根据实际情况进行配置；根据日志中消息类型标记的不同进行不同的处理：

(1)点击或搜索日志类消息，按照能源用户或能源服务商、能源商品类型或能源需求类型统计在该批次消息中能源用户点击和搜索各类能源商品的次数及能源服务商点击和搜索各类能源需求的次数；根据统计结果计算能源用户当前最感兴趣的能源商品类型，再根据能源用户对该能源商品类型的评分、能源用户平均消费金额及各个能源商品综合评分、价格，计算出能源用户当前可能最感兴趣的能源商品；同时将本批次统计的次数累加到历史总次数中，基于历史总次数再计算出能源用户历史最感兴趣的能源商品，基于用户标签计算出能源用户可能感兴趣的能源商品；对于能源服务商点击和搜索各类能源需求的次数只累加到历史总次数中，用于能源需求推荐；

(2)合同类消息，按照能源用户或能源服务商、能源商品类型或能源需求类型统计该批次消息中能源用户对各类能源商品生成合同的金额及能源服务商对各类能源需求生成合同的次数，用于计算能源用户对各类能源品的平均消费金额、评分和能源服务商在各类能源需求的平均得分；

(3)能源需求类消息，根据能源服务商对各类需求点击和搜索次数，计算出对该类能源需求最感兴趣的能源服务商，再根据能源服务商在该类能源需求的得分筛选出推荐结果。

而且，步骤四的数据存储部分采用Oracle关系型数据库，主要存储内容包括：能源用户信息、能源服务商信息、商品信息、合同信息、日志信息及推荐结果信息。

为了更清楚地说明本发明的具体实施方式，下面提供一种实施例：

本发明数据流图如图1所示，具体步骤如下：

(1)前端日志采集程序记录用户的点击或搜索日志、生成合同日志、发布能源需求日志，并将日志以特定的JSON格式发送到Kafka消息中间件中的特定Topic。

(2)Kafka接收日志消息，并按照配置的分区和副本策略保存消息。

(3)Spark Streaming实时计算程序启动，首先读取计算所需各种数据，包括用户历史操作日志、能源商品信息、企业对应的标签信息、标签对应的商品类型、用户名称对应企业信息，并在读取完成后将数据处理成需要的格式。

(4)Spark Streaming实时计算程序连接Kafka，每隔特定时长拉取用户实时日志消息，并根据日志消息类型作相应处理做相应处理。

(5)根据点击或搜索型日志消息、历史统计结果、用户标签产生用户当前最感兴趣、历史最感兴趣及可能最感兴趣的能源商品推荐结果。

(6)根据合同型日志消息及点击或搜索型日志消息累加历史统计结果。

(7)根据能源需求型日志消息及历史统计结果产生能源需求推荐结果。

(8)Spark Streaming计算程序将产生的推荐结果写入Oracle数据库。

(9)前端页面拉取推荐结果并展示。

本实施样例为一个单独运行程序，该程序采用流批融合的分布式计算实现能源商品及需求的精准推送，其输出结果既有基于流式计算产生的用户当前最感兴趣的推荐数据，又有基于批量计算产生的用户综合最感兴趣的推荐数据。该程序通过流批融合克服了现有推荐系统的缺点：独立采用流式计算实现的推荐系统，其计算结果基于实时数据，存在片面性；独立采用批量计算实现的推荐系统，其计算结果基于过去较长一段时间内的批量数据，存在滞后性。

程序主体如下所示：

生产环境运行如下所示：

本系统初始化时与消息中间件Kafka建立实时数据流，同时会一次性读取目前已存在数据库的各类历史批量数据存入内存。

系统正常运行后，不断获取实时流数据，实时流数据进入流批融合数据处理流程，流批融合数据处理流程一是对实时流数据进行清洗、转换、提取、计算，基于实时流数据产生用户当前最感兴趣的推荐数据，二是利用自定义累加器将实时流数据与全量历史数据累加，基于批量数据计算相关统计指标并产生用户综合最感兴趣的推荐数据。

流批融合数据处理程序如下所示：

批量数据累加器如下所示：

处理流数据的同时累加数据并进行批量计算如下所示：

最后，本发明的未述之处均采用现有技术中的成熟产品及成熟技术手段。

应当理解的是，对本领域普通技术人员来说，可以根据上述说明加以改进或变换，而所有这些改进和变换都应属于本发明所附权利要求的保护范围。