基于实时计算的基数估计的方法和系统

摘要

本发明提供一种基于实时计算的基数估计的方法和系统，能够基于概率和统计理论进行高效的基数估计计算，从而满足大数据场景的实时基数计算需求。该方法包括在Storm系统中的执行下列步骤：实时获取日志消息；解析所述日志消息以获取指标信息，所述指标信息包括各指标的名称及对应的指标值；利用HLL基数估计算法对各指标进行基数估计；输出各指标的基数。

说明书

技术领域

本发明涉及计算机技术及软件领域，尤其涉及一种基于实时计算的基数估计的方法和系统。

背景技术

基数计数是对一个可重复集合中不重复元素个数的计算。例如计算整个网站或店铺的独立访客等。在大数据的背景下，传统的基数计算方法遇到一些困难，主要表现在随着数据量和分析维度的增加，所需的计算资源和存储资源迅速膨胀。因此需要一种高效的基数估计机制。

基数估计算法是一类概率算法，可以在误差可控的前提下以远低于精确计算的时间和空间消耗对基数进行估计。算法特点:1.误差可控2.时间和空间复度仅与估计值标准差及基数上限有关3.可合并。现有的基数估计计算通常采用Redis的HyperLogLog Counting功能，完成对基数的估计计数。其中，

但是利用Redis的HyperLogLog Counting进行基数估计计算仍然存在如下缺点：Redis并没有实现基数算法的可合并特性，导致大数据量下处理能力不可扩展；由于将整个计算环节交给Redis进行处理，系统和Redis形成强依赖关系；此外，搭建Redis集群也产生较大的运营成本。

发明内容

有鉴于此，本发明提供一种基于实时计算的基数估计的方法和系统，能够基于概率和统计理论进行高效的基数估计计算，从而满足大数据场景的实时基数计算需求。

为实现上述目的，根据本发明的一个方面，提供了一种基于实时计算的基数估计的方法。

本发明的基于实时计算的基数估计的方法包括在Storm系统中的执行下列步骤：实时获取日志消息；解析所述日志消息以获取指标信息，所述指标信息包括各指标的名称及对应的指标值；利用HLL基数估计算法对各指标进行基数估计；输出各指标的基数。

可选地，所述方法还包括：解析所述日志消息之后，对获取的指标信息进行校验，以删除异常指标信息。

可选地，利用HLL基数估计算法对各指标进行基数估计还包括：将所述指标信息随机分配至基数计算层的多个线程，各线程根据分配到的指标信息，为各指标创建HLL对象，利用HLL基数估计算法中的Offer方法将各指标的指标值加入到对应的HLL对象中，然后定时将HLL对象发送到基数集合合并层；以及所述基数集合合并层接收HLL对象，并按各HLL对象的指标名称创建各指标的总HLL对象，然后利用HLL算法中的Merge方法将HLL对象按照指标名称合并到各自对应的总HLL对象中，以及定时利用HLL算法中的Cardinality方法对各指标的总HLL对象进行计数，以获得各指标的基数。

可选地，所述方法还包括：所述HLL对象及所述总HLL对象均保存在位于其所在服务器内存中的LRUmap中。

可选地，所述方法还包括：定时将总HLL对象保存至外部的数据库。

可选地，所述方法还包括：定时将各指标的基数保存到外部的数据库。

可选地，所述方法中的定时是指：记录上次操作的时间，若当前时刻与上次操作的时间差小于预设阈值，则不进行相应操作，若当前时刻与上次操作的时间差大于预设阈值，则进行相应操作。

为实现上述目的，根据本发明的另一个方面，提供了一种基于实时计算的基数估计的系统。

本发明的基于实时计算的基数估计的系统包括：存储器和处理器，其中，所述存储器存储指令；所述处理器执行所述指令用于：在Storm系统中的执行下列步骤：实时获取日志消息；解析所述日志消息以获取指标信息，所述指标信息包括各指标的名称及对应的指标值；利用HLL基数估计算法对各指标进行基数估计；输出各指标的基数。

可选地，所述处理器还用于：解析所述日志消息之后，对获取的指标信息进行校验，以删除异常指标信息。

可选地，所述处理器还用于：将所述指标信息随机分配至基数估计计算层的多个线程，各线程根据分配到的指标信息，为各指标创建HLL对象，利用HLL基数估计算法中的Offer方法将各指标的指标值加入到对应的HLL对象中，然后定时将HLL对象发送到基数集合合并层；以及所述基数集合合并层接收HLL对象，并按各HLL对象的指标名称创建各指标的总HLL对象，然后利用HLL算法中的Merge方法将HLL对象按照指标名称合并到各自对应的总HLL对象中，以及定时利用HLL算法中的Cardinality方法对各指标的总HLL对象进行计数，以获得各指标的基数。

可选地，所述处理器还用于：将所述HLL对象及所述总HLL对象均保存在位于其所在服务器内存中的LRUmap中。

可选地，所述处理器还用于：定时将总HLL对象保存至外部的数据库。

可选地，所述处理器还用于：定时将各指标的基数保存到外部的数据库。

可选地，所述处理器还用于：其中的定时是指记录上次操作的时间，若当前时刻与上次操作的时间差小于预设阈值，则不进行相应操作，若当前时刻与上次操作的时间差大于预设阈值，则进行相应操作。

根据本发明的技术方案，通过利用实时计算系统Storm可水平扩容、容灾等机制以及基数估计算法HLL的低存储空间、集合可合并等特性的结合，从而可以保证在扩容方便、占用存储空间少的前提下，实现对大数据环境下的基数进行实时高效的计数；通过在解析日志消息后对获取的数据进行校验，从而可以保证计算的精确性，避免计算的浪费；通过将HLL对象及总HLL对象均保存在位于其所在服务器内存中的LRUmap中，从而可以避免因长期运行而占用过大内存的现象的发生；通过定期将各指标的总HLL对象保存至Storm系统外部的数据库，从而可以保证在系统宕机或任务的部分节点重启时，从数据库中恢复中间结果；通过定时将各指标的基数保存到Storm系统外部的数据库，从而可以保证对基数计算结果保存以及实时统计和呈现；通过设置在内存中记录Bolt中上次的操作时间，并进行时间差比较的定时机制，而非在Storm系统中为“定时”另启线程维护定时，从而可以降低程序的复杂度。

附图说明

附图用于更好地理解本发明，不构成对本发明的不当限定。其中：

图1是根据本发明实施例的基于实时计算的基数估计的方法的主要步骤的示意图；

图2是根据本发明实施例的基于实时计算的基数估计的方法的主要流程的示意图；

图3是根据本发明实施例的基于实时计算的基数估计的系统的主要部分的示意图。

具体实施方式

以下结合附图对本发明的示范性实施例做出说明，其中包括本发明实施例的各种细节以助于理解，应当将它们认为仅仅是示范性的。因此，本领域普通技术人员应当认识到，可以对这里描述的实施例做出各种改变和修改，而不会背离本发明的范围和精神。同样，为了清楚和简明，以下的描述中省略了对公知功能和结构的描述。

图1是根据本发明实施例的基于实时计算的基数估计的方法的主要步骤的示意图。

如图1所示，本发明实施例的基于实时计算的基数估计的方法主要包括在Storm系统中的执行下列步骤：

步骤S11：实时获取日志消息。基数计数是实际应用中一种常见的计算场景，它是对一个可重复集合中不重复元素个数的统计。实际应用中可以用作计算各类指标的基数，例如在电子商务领域，可以计算整个网站当天“独立访客数(Unique Visitors，即UV)”和“浏览商品SKU种类数(即SKUSum)”。本发明实施例中以此举例说明具体计算过程。

随着数据量和分析维度的增加，所需的计算资源和存储资源迅速增多。在计算的过程中，待计数的指标可能有多种数据来源。例如，在上述统计一个网站的当天的“UV”的时候，可能需要获取来自计算机端的访问数据，同时还需要获取来自移动终端的访问数据，因此，首先，要完成对底层数据日志的抓取。但是通过不同来源获取的数据的数据格式可能不统一，因此还需要将抓取的数据还原成统一的格式。获取数据后，将数据放入到消息传输队列，例如可以是Kafka中。此处并不限于Kafka消息传输队列，可以使用其他消息传输队列，只要能实现在消息的传输过程中保存消息的作用即可。

在将数据传输到消息队列后，为了便于实时对某些特定指标进行基数计数，本发明引入分布式的、容错的实时计算系统Storm(Storm为分布式实时计算提供了一组通用原语，可被用于“流处理”之中，实时处理消息并更新数据库。这是管理队列及工作者集群的另一种方式。Storm也可被用于“连续计算”，对数据流做连续查询，在计算时就将结果以流的形式输出给用户。它还可被用于“分布式RPC”，以并行的方式运行昂贵的运算)进行流式并行计算。首先利用Storm系统进行日志消息的实时获取。即利用spout从消息传输队列(例如可以是kafka)中接收消息，并对压缩的消息进行解压，然后转换成对应的Tuple，最后将结果Tuple随机分发到下层Bolt中。

在本步骤的实时获取日志消息之后，从步骤S 12进行处理。

步骤S12：解析所述日志消息以获取指标信息，所述指标信息包括各指标的名称及对应的指标值。

在步骤S11实时获取日志消息后，将日志消息分配至数据解析层(在Storm系统中，称为DataAnanlyzeBolt)进行处理。数据分配时可以有多种分配方式，本发明实施例中，为了保证数据被平均的分散到各个数据解析层的task中，是利用随机分组Shuffle Grouping方式进行，以便数据解析的压力被分散开。本步骤要分别解析出日志消息中对应的待计数指标的信息。

以实时计算电商网站当天的“UV”和“浏览商品SKU种类数”为例：要解析出的业务ID是日志所产生的时间，指标名称是用户ID(对应的指标基数称为“UV”)和访问商品SKU(对应的指标基数称为“SKUSum”)，指标名称对应的指标值分别是用户设备的UUID和访问的商品的SKUID。然后对获取的指标信息进行业务校验：当存在日志中没有对应的UUID或SKUID或者获取的UUID、SKUID格式不正确等情况时，则不向下一层发送对应的Tuple，从而删除异常格式的数据。最后将组成[业务ID＝ID值，指标名称＝指标值]，例如[业务ID＝日志所产生时间，用户ID＝UUID]和/或[业务ID＝日志所产生时间，浏览商品SKU＝SKUID]的Tuple发送到下一层。

步骤S13：利用HLL基数估计算法对各指标进行基数估计。本发明实施例中该步骤可以是在Storm的数据基数估计计算层DataHLLCaluBolt以及数据基数集合合并层DataHLLMergeBolt完成，主要包括：将所述指标信息随机分配至基数估计计算层的多个线程，各线程根据分配到的指标信息，为各指标创建HLL对象，利用HLL基数估计算法(HyperLogLog Counting：是用来做基数统计的算法，简称HLL。HLL的优点是，在输入元素的数量或者体积非常非常大时，计算基数所需的空间总是固定的)中的Offer方法将各指标的指标值加入到对应的HLL对象中，然后定时将HLL对象发送到基数集合合并层；以及所述基数集合合并层接收HLL对象，并按各HLL对象的指标名称创建各指标的总HLL对象，然后利用HLL算法中的Merge方法将HLL对象按照指标名称合并到各自对应的总HLL对象中，以及定时利用HLL算法中的Cardinality方法对各指标的总HLL对象进行计数，以获得各指标的基数。

具体而言，在数据基数估计计算层DataHLLCaluBolt要解决的问题：将步骤S12解析得到的大量需要做去重的指标数据平均分散到各个task当中，从而使得计算的压力被水平分散开，然后利用HLL基数估计算法中的Offer方法进行各自的去重操作。以下分别介绍下task接收数据的方式、task基数计算的过程和组装给下游的Tuple：

1.按照Shuffle Grouping方式，接收数据解析层的Tuple，这样可以保证数据被平均分配到task中；

2.当task拿到对应的Tuple，此处tuple中的数据格式为[业务ID＝ID值，指标名称＝指标值]，根据“业务ID+指标名称”为每个业务ID相应的指标创建基数估计HLL对象，并将指标值通过HLL算法中的Offer方法(Offer方法是将加入的指标值进行哈希散列，并进行估计计算，且计算结果对象的内存使用大小不会发生变化)加入到HLL对象内。这里创建的HLL对象保存在task初始化在内存中创建的LRUmap当中(LRUmap是在有限的集合里面，如果存储的时候，数据超出了限制，那么就淘汰最近未使用的数据，保证了对象有释放的可能，不会因为长期运行而把内存占用满。在本发明实施例中，DataHLLCaluBolt初始化过程创建一个有限的LRUmap集合，它用来存放指标信息去重后的要向下一层发送的这一批次内的HLL对象。当下述定时触发向下层发送消息完成后，则将对LRUmap集合清空。这里利用LRUmap的“数据超出了集合限制，那么就淘汰最近未使用的数据”这一特性，确保了在一个批次内的对象不会因为创建太多，导致内存过大出现问题)；

3.定时(此时间可根据消息量和模块的并行度进行配置预设阈值，默认1秒，间隔是让每给HLL对象能够积累一些基数，这样可以大大减少传递到下一层的数据量)把HLL对象序列化后发向下一层。发送的Tuple格式为：[业务ID，指标名称，HLL序列化后的对象]。

在数据基数集合合并层DataHLLMergeBolt要解决的问题：一是将上一层分散在不同task中的HLL对象按照“业务ID+指标名称”的方式进行合并，然后利用HLL算法中的Cardinality方法对合并后的HLL对象计算最终的估计值，并定时发送到下层。二是要实现容灾机制，在任务出现重启时不会丢失计算值。下面详细解决过程：

1.本层采用上层Tuple中的“业务ID+指标名称”进行对task分组，如同一天的访问用户的ID数据将分配到同一个task当中，同一天的访问商品SKU数据将分配到另一个task当中，这样做是为了保证上层不同的指标能在同一个task中进行合并处理。

2.接收到相应的Tuple后，首先将上层的传入的HLL对象进行反序列化还原成可用对象，然后根据“业务ID+指标名称”为各指标创建总HLL对象(同DataHLLCaluBolt一样，这里创建的总HLL对象保存在task初始化在内存中创建的LRUmap当中。与DataHLLCaluBolt层不同的是，此处LRUmap集合内的总HLL对象的删除机制，并没有像DataHLLCaluBolt那样主动去清除，而是直接利用LRUmap集合本身自带的“数据超出了集合限制，那么就淘汰最近未使用的数据”功能来限制集合的大小)。其中，创建总HLL对象的过程如下：当上层数据到达DataHLLMergeBolt后，首先按照消息的业务维度在内存中的LRUmap中查找是否已经存在此维度总HLL对象，如果存在则直接和上层数据进行合并，否则从外部数据库，例如可以是HBase(即下文所述用于存放总HLL对象的外部数据库)当中查找是否存在此对象，如果HBase中存在则将其加载到内存中的LRUmap内，并进行合并，不存在则创建新的此维度总HLL对象，并且和上层对象进行合并后放入LRUmap当中(这也是为什么该层LRUmap集合中的总HLL对象的删除机制不能采用每个批次都主动清空的原因：如果每个批次都清空，则上层数据在进入合并层后，判断LRUmap中没有相应的总HLL对象就会去HBase中确认，这样会对HBase造成一定的压力)。再利用HLL基数估计算法中的Merge方法(Merge：将多个并行运算的HLL对象进行合并)把上层的HLL对象合并到总HLL对象上来。

3.最后定时(此时间可根据消息量和业务对计算指标的及时性要求进行决定预设阈值，可默认1秒，这样可以大大减少传递到下一步的数据量)利用HLL算法中的cardinality方法(cardinality方法是对HLL对象进行统计，返回最终结果值)对总HLL对象进行统计，得到最终的指标基数结果值(例如本发明实施例中举例的UV及SKUSum)，并组装成[业务ID＝ID号，指标基数＝指标基数估计结果]的Tuple发向下一层Bolt。

4.容灾机制：计算过程中，将总HLL对象定时(此预设阈值可根据消息量和外部数据库的负载能力决定，默认1秒)持久化到Storm系统的外部数据库(例如但不限于可以是HBase)当中，当task重启时可以先从HBase加载相应的对象。这种方式可以保证数据在宕机或任务的部分节点重启时，可以从外部数据库中恢复中间结果，这里用到的HBase表需要设置过期失效，以保证表里的数据量可控。

步骤S14：输出各指标的基数。此步骤是在Storm系统的存储层PersistBolt完成业务结果存储。经过基数估计后，本层task可以进行分组(例如文中例子中的根据“业务ID”)接收DataHLLMergeBolt的估计结果，接收的内容格式为[业务ID＝ID号，指标基数＝指标基数估计结果]，最终指标基数结果值定时(此预设阈值可根据消息量和业务对计算指标的及时性要求进行决定，默认1秒，这样做是为了减少与数据库的交互请求)存入Storm系统的外部数据库(例如但不限于可以是HBase)的业务结果表中。这里按照业务分组接收上层结果，是为了保证同样的业务ID(例如本发明实施例中的“当天”)被同一个task进行处理，以便在同一时间更新HBase业务表中的对应记录的task只有一条。

在本发明实施例的基于实时计算的基数估计的方法中，各Bolt中“定时”的机制是指：在内存中记录Bolt中上次进行相应操作的时间，若当前时刻与上次操作的时间差小于预设阈值(例如可以是1S)，则不进行相应操作，若当前时刻与上次操作的时间差大于预设阈值(例如可以是1S)，则进行相应操作。这种方式的好处是：不用在Storm系统中为“定时”另启线程维护定时机制，降低了程序的复杂度。本发明实施例不限于上述“定时”机制，可根据数据量的大小或系统需求，更换其他定时器装置。

上述步骤的具体操作流程见图2。

根据本发明实施例的基于实时计算的基数估计的方法可以看出，通过利用实时计算系统Storm可水平扩容、容灾等机制以及基数估计算法HLL的低存储空间、集合可合并等特性的结合，从而可以保证在扩容方便、占用存储空间少的前提下，实现在大数据环境下对基数进行实时高效的计数；通过在解析日志消息后对获取的数据进行校验，从而可以保证计算的精确性，避免计算的浪费；通过将HLL对象及总HLL对象均保存在位于其所在服务器内存中的LRUmap中，从而可以避免因长期运行而占用过大内存的现象的发生；通过定期将生成各指标的总HLL对象保存至Storm系统外部的数据库，从而可以保证在系统宕机或任务的部分节点重启时，从数据库中恢复中间结果；通过定期将各指标的基数保存到Storm系统外部的数据库，从而可以保证对基数计算结果保存以及实时统计和呈现；通过设置在内存中记录Bolt中上次的操作时间，并进行时间差比较的定时机制，而非在Storm系统中为“定时”另启线程维护定时，从而可以降低程序的复杂度。

图3是根据本发明实施例的基于实时计算的基数估计的系统的主要部分的示意图。

如图3所示，本发明实施例的基于实时计算的基数估计的系统30主要包括如下部分：存储器301和处理器302。

其中，存储器301存储指令；处理器302执行所述指令用于：在Storm系统中的执行下列步骤：实时获取日志消息；解析所述日志消息以获取指标信息，所述指标信息包括各指标的名称及对应的指标值；利用HLL基数估计算法对各指标进行基数估计；输出各指标的基数。

所述处理器302还可用于：解析所述日志消息之后，对获取的指标信息进行校验，以删除异常指标信息。

所述处理器302还可用于：将所述指标信息随机分配至基数估计计算层的多个线程，各线程根据分配到的指标信息，为各指标创建HLL对象，利用HLL基数估计算法中的Offer方法将各指标的指标值加入到对应的HLL对象中，然后定时将HLL对象发送到基数集合合并层；以及所述基数集合合并层接收HLL对象，并按各HLL对象的指标名称创建各指标的总HLL对象，然后利用HLL算法中的Merge方法将HLL对象按照指标名称合并到各自对应的总HLL对象中，以及定时利用HLL算法中的Cardinality方法对各指标的总HLL对象进行计数，以获得各指标的基数。

所述处理器302还可用于：所述HLL对象及所述总HLL对象均保存在位于其所在服务器内存中的LRUmap中。

所述处理器302还可用于：定时将总HLL对象保存至外部的数据库。

所述处理器302还可用于：定时将各指标的基数保存到外部的数据库。

所述处理器302还可用于：其中的定时是指记录上次操作的时间，若当前时刻与上次操作的时间差小于预设阈值，则不进行相应操作，若当前时刻与上次操作的时间差大于预设阈值，则进行相应操作。

从以上描述可以看出，通过利用实时计算系统Storm可水平扩容、容灾等机制以及基数估计算法HLL的低存储空间、集合可合并等特性的结合，从而可以保证在扩容方便、占用存储空间少的前提下，实现在大数据环境下对基数进行实时高效的计数；通过在解析日志消息后对获取的数据进行校验，从而可以保证计算的精确性，避免计算的浪费；通过将HLL对象及总HLL对象均保存在位于其所在服务器内存中的LRUmap中，从而可以避免因长期运行而占用过大内存的现象的发生；通过定期将生成各指标的总HLL对象保存至Storm系统外部的数据库，从而可以保证在系统宕机或任务的部分节点重启时，从数据库中恢复中间结果；通过定期将各指标的基数保存到Storm系统外部的数据库，从而可以保证对基数计算结果保存以及实时统计和呈现；通过设置在内存中记录Bolt中上次的操作时间，并进行时间差比较的定时机制，而非在Storm系统中为“定时”另启线程维护定时，从而可以降低程序的复杂度。

上述具体实施方式，并不构成对本发明保护范围的限制。本领域技术人员应该明白的是，取决于设计要求和其他因素，可以发生各种各样的修改、组合、子组合和替代。任何在本发明的精神和原则之内所作的修改、等同替换和改进等，均应包含在本发明保护范围之内。