一种指标实时分析系统及指标实时计算方法

摘要

本发明提供一种指标实时分析系统及指标实时计算方法，该指标实时分析系统包括：日志接入模块、实时计算引擎模块和存储模块；所述日志接入模块，用于采集并解析原始日志数据，得到解析后的日志数据作为流数据，并将所述流数据写入所述存储模块；所述实时计算引擎模块，用于从所述存储模块获取所述流数据，对所述流数据进行计算，得到指标数据，并将所述指标数据写入所述存储模块。通过上述方式，本发明能够加快数据处理的速度，保证指标计算的实时性，做到即时输出。

说明书

技术领域

本发明涉及大数据处理技术领域，尤其涉及一种指标实时分析系统及指标实时计算方法。

背景技术

随着互联网的持续发展，企业需要及时对大量线上产品和海量的用户访问日志进行监控，以实时有效地掌控产品的使用情况，以及快速客观地评估产品运行状况。

现有技术中，首先将采集到的日志进行解析，保存到搜索引擎之后，再对搜索引擎中的数据进行统计分析，获取业务指标数据，从而进行监控和告警。具体的，将采集到的日志解析，保存到搜索引擎Elasticsearch，再查询Elasticsearch里解析之后的数据进行计算得出指标，环节比较多，在海量日志的场景下，延迟比较严重。举例说明，刷卡业务系统的每一笔交易都会产生一条日志，包含交易时间、交易终端和交易金额等信息。如果要监控每分钟的交易金额，那就需要查询解析后的日志，提取其中的交易金额，按照分钟进行累加，得出一系列指标数值，从而可以展示计算出的指标，例如在监控界面以图表形式展示。而由于指标的计算是基于查询Elasticsearch实现的，随着指标数量的增长，对Elasticsearch的查询频率会非常高。另一方面，Elasticsearch还需要保存日志数据，每秒钟都有大量的数据需要落盘，同时还要构建索引以便查询，Elasticsearch压力会非常大，处于不稳定状态，而且还会影响系统的吞吐量，造成数据积压。

因此，如何实时计算指标，不积压数据，是目前亟待解决的技术问题。

发明内容

有鉴于此，本发明提供一种指标实时分析系统及指标实时计算方法，用于解决现有技术中指标计算不够实时及数据积压的问题。

为解决上述技术问题，第一方面，本发明提供一种指标实时分析系统，包括：日志接入模块、实时计算引擎模块和存储模块；

所述日志接入模块，用于采集并解析原始日志数据，得到解析后的日志数据作为流数据，并将所述流数据写入所述存储模块；

所述实时计算引擎模块，用于从所述存储模块获取所述流数据，对所述流数据进行计算，得到指标数据，并将所述指标数据写入所述存储模块。

优选的，所述日志接入模块采用至少一个logstash组件采集并解析所述原始日志数据。

优选的，所述存储模块存储有日志数据对应的多个日志主题；

所述日志接入模块，用于根据所述流数据的日志属性，确定所述流数据对应的日志主题，将所述流数据存储至对应的日志主题中；

或者，

所述存储模块存储有日志数据对应的一个日志主题；

所述日志接入模块，用于根据所述流数据的日志属性，为所述流数据打上对应的标签，并将打上标签后的所述流数据存储至所述日志主题中。

优选的，所述存储模块存储有指标数据对应的多个指标主题；

所述实时计算引擎模块，还用于根据所述指标数据的属性，确定所述指标数据对应的指标主题，将所述指标数据存储至对应指标主题中。

优选的，所述存储模块包括至少两个第一数据库；

所述存储模块，还用于将所述指标数据同时存储至所述至少两个第一数据库中。

优选的，所述存储模块，用于将第一预设时间之前的指标数据存储至第二数据库，将第一预设时间之后的指标数据存储于第一数据库。

优选的，所述第一数据库为Prometheus，所述第二数据库为OpenTSDB。

优选的，所述存储模块采用Kafka集群进行存储。

优选的，所述指标实时分析系统还包括：

查询模块，还用于根据用户的输入，产生指标查询命令；

展示模块，用于根据所述指标查询命令，查询所述存储模块中存储的所述指标数据，并展示查询到的所述指标数据。

优选的，所述实时计算引擎模块，用于根据指标的配置信息，生成与所述指标对应的实时计算任务，采用所述实时计算任务对所述流数据进行计算，得到所述指标数据，并将所述指标数据写入所述存储模块。

优选的，所述指标实时分析系统还包括：

配置中心，与所述实时计算引擎模块连接，用于存储所述指标的配置信息，所述指标的配置信息包括以下至少之一：所述指标的名称、所述指标的状态、所述指标的描述、所述指标的类型、所述指标的指标值类型、所述指标的标签和所述指标的计算定义；

所述实时计算引擎模块，还用于获取所述指标的配置信息，对所述指标的配置信息进行解析，根据解析结果，生成所述实时计算任务，并运行所述实时计算任务，所述实时计算引擎模块中运行有至少一个实时计算任务。

优选的，所述配置中心采用Zookeeper组件实现。

优选的，所述指标实时分析系统还包括：

指标配置模块，用于显示新建或更新所述指标的用户界面，并接收用户的新建或更新所述指标的操作，根据所述操作，新建或更新所述指标的配置信息，并存储至所述配置中心。

优选的，所述实时计算引擎模块，用于针对每一所述实时计算任务，生成用于监听指标变化的注册命令，并提交至所述配置中心；

所述配置中心，用于当新建指标或更新指标时，生成携带所述新建指标或更新指标的配置信息的通知信息，向所述实时计算引擎模块中的每一所述实时计算任务广播所述通知信息。

优选的，所述指标的配置信息包括所述指标的计算定义；

所述实时计算引擎模块，用于解析所述指标的计算定义，得到解析结果，所述解析结果中包括至少一个函数，其中，每一函数对应至少一个算子，将所述至少一个函数转换为至少一个算子，根据所述至少一个算子生成所述实时计算任务，所述实时计算任务包括：算子和算子之间的连接关系以及各个算子的配置。

优选的，所述指标的计算定义采用SQL语言描述。

优选的，所述指标的计算定义包括以下函数中的至少之一：SELECT、FROM、JOIN、WHERE、GROUP BY、HAVING、ORDER BY；

所述算子包括以下至少之一：排序算子、表名匹配算子、关联表处理算子、WHERE条件匹配算子、聚合函数检测算子、聚合键生成算子、窗口检测算子、窗口计算算子、聚合计算算子、HAVING处理算子和字段选取算子。

优选的，所述指标实时分析系统还包括：

函数自定义模块，用于获取自定义的函数，将自定义的函数注册至SQL解析的上下文中。

优选的，所述实时计算任务中的算子包括多个实例，至少两个所述实例运行在不同处理节点上；

所述实时计算引擎模块，用于从预设时间段内的所述流数据中抽取一数据样本，获取所述数据样本在所述不同处理节点上的数据分布比例；根据所述数据样本中的数据分布比例，分配之后时间段的在不同处理节点上的数据分布。

优选的，所述实时计算引擎模块，用于获取所述数据样本中的流数据对应的键值，若一键值对应的流数据的数量大于或等于第一预设数量阈值，在所述一键值上增加一随机标识，以将所述一键值对应的流数据分配至不同处理节点；去除经所述不同处理节点处理后的流数据中的所述随机标识，合并去除所述随机标识后的流数据。

优选的，所述实时计算引擎模块，用于将流数据与静态数据进行关联时，获取所述流数据对应的预设键值的数量和所述静态数据对应的预设键值的数量，比较所述流数据对应的预设键值的数量与所述静态数据对应的预设键值的数量；

若所述流数据对应的预设键值的数量大于所述静态数据对应的预设键值的数量，向所述流数据广播所述静态数据对应的预设键值；

若所述流数据对应的预设键值的数量小于或等于所述静态数据对应的预设键值的数量，向所述静态数据广播所述流数据对应的预设键值；

关联所述流数据与所述静态数据。

第二方面，本发明还提供一种指标实时计算方法，包括：

采集并解析原始日志数据，得到解析后的日志数据作为流数据，并存储所述流数据；

获取所述流数据，对所述流数据进行计算，得到指标数据，并存储所述指标数据。

优选的，所述存储所述流数据的步骤包括：

根据所述流数据的日志属性，确定所述流数据对应的日志主题，将所述流数据存储至对应的日志主题中，所述日志主题的个数为多个；

或者，

根据所述流数据的日志属性，为所述流数据打上对应的标签，并将打上标签后的所述流数据存储至一个日志主题中。

优选的，所述存储所述指标数据的步骤包括：

根据所述指标数据的属性，确定所述指标数据对应的指标主题，将所述指标数据存储至对应指标主题中。

优选的，所述存储所述指标数据的步骤包括：

将所述指标数据同时存储至至少两个第一数据库中。

优选的，所述存储所述指标数据的步骤包括：

将第一预设时间之前的指标数据存储至第二数据库，将第一预设时间之后的指标数据存储于第一数据库。

优选的，所述第一数据库为Prometheus，所述第二数据库为OpenTSDB。

优选的，所述指标实时计算方法还包括：

根据用户的输入，产生指标查询命令；

根据所述指标查询命令，查询所述指标数据，并展示查询到的所述指标数据。

优选的，所述对所述流数据进行计算，得到指标数据的步骤包括：

根据指标的配置信息，生成与所述指标对应的实时计算任务，采用所述实时计算任务对所述流数据进行计算，得到所述指标数据。

优选的，所述指标实时计算方法还包括：

存储所述指标的配置信息，所述指标的配置信息包括以下至少之一：所述指标的名称、所述指标的状态、所述指标的描述、所述指标的类型、所述指标的指标值类型、所述指标的标签和所述指标的计算定义；

获取所述指标的配置信息，对所述指标的配置信息进行解析，根据解析结果，生成所述实时计算任务，并运行所述实时计算任务。

优选的，所述指标实时计算方法还包括：

显示新建或更新所述指标的用户界面，并接收用户的新建或更新所述指标的操作，根据所述操作，新建或更新所述指标的配置信息，并存储新建或更新的指标的配置信息。

优选的，所述指标实时计算方法还包括：

针对每一所述实时计算任务，生成用于监听指标变化的注册命令；

当新建指标或更新指标时，生成携带所述新建指标或更新指标的配置信息的通知信息，向每一所述实时计算任务广播所述通知信息。

优选的，所述指标的配置信息包括所述指标的计算定义，所述指标实时计算方法还包括：

解析所述指标的计算定义，得到解析结果，所述解析结果中包括至少一个函数，其中，每一函数对应至少一个算子，将所述至少一个函数转换为至少一个算子，根据所述至少一个算子生成所述实时计算任务，所述实时计算任务包括：算子和算子之间的连接关系以及各个算子的配置。

优选的，所述指标的计算定义采用SQL语言描述。

优选的，所述指标的计算定义包括以下函数中的至少之一：SELECT、FROM、JOIN、WHERE、GROUP BY、HAVING、ORDER BY；

所述算子包括以下至少之一：排序算子、表名匹配算子、关联表处理算子、WHERE条件匹配算子、聚合函数检测算子、聚合键生成算子、窗口检测算子、窗口计算算子、聚合计算算子、HAVING处理算子和字段选取算子。

优选的，所述指标实时计算方法还包括：

获取自定义的函数，将自定义的函数注册至SQL解析的上下文中。

优选的，所述实时计算任务中的算子包括多个实例，至少两个所述实例运行在不同处理节点上；

所述指标实时计算方法还包括：从预设时间段内的所述流数据中抽取一数据样本，获取所述数据样本在所述不同处理节点上的数据分布比例；根据所述数据样本中的数据分布比例，分配之后时间段的在不同处理节点上的数据分布。

优选的，所述指标实时计算方法还包括：

获取所述数据样本中的流数据对应的键值，若一键值对应的流数据的数量大于或等于第一预设数量阈值，在所述一键值上增加一随机标识，以将所述一键值对应的流数据分配至不同处理节点；

去除经所述不同处理节点处理后的流数据中的所述随机标识，合并去除所述随机标识后的流数据。

优选的，所述指标实时计算方法还包括：

将流数据与静态数据进行关联时，获取所述流数据对应的预设键值的数量和所述静态数据对应的预设键值的数量；

比较所述流数据对应的预设键值的数量与所述静态数据对应的预设键值的数量；

若所述流数据对应的预设键值的数量大于所述静态数据对应的预设键值的数量，向所述流数据广播所述静态数据对应的预设键值；

若所述流数据对应的预设键值的数量小于或等于所述静态数据对应的预设键值的数量，向所述静态数据广播所述流数据对应的预设键值；

关联所述流数据与所述静态数据。

第三方面，本发明还提供一种指标实时分析系统，包括存储器、处理器及存储在所述存储器上并可在所述处理器上运行的计算机程序；所述处理器执行所述计算机程序时实现上述的指标实时计算方法。

第四方面，本发明还提供一种计算机可读存储介质，其上存储有计算机程序，该计算机程序被处理器执行时实现上述指标实时计算方法中的步骤。

本发明的上述技术方案的有益效果如下：区别于现有技术的情况，本发明通过日志接入模块对原始日志进行解析，并将解析后的数据写入存储模块，实时计算引擎模块从存储模块中获取解析后的日志数据，计算得出指标数据，将计算好的指标数据，输出到存储模块中，能够加快数据处理的速度，保证指标计算的实时性，做到即时输出。

附图说明

图1为本发明实施例一的指标实时分析系统的结构示意图；

图2为本发明实施例的配置指标界面的显示示意图；

图3为本发明实施例的配置中心与实时计算引擎模块的数据交互的流程示意图；

图4为本发明实施例的算子实时计算的流程示意图；

图5为本发明的一具体应用场景的指标实时分析系统的结构示意图；

图6为本发明实施例二的指标实时计算方法的流程示意图；

图7为本发明实施例三的指标实时分析系统的结构示意图。

具体实施方式

为使本发明实施例的目的、技术方案和优点更加清楚，下面将结合本发明实施例的附图，对本发明实施例的技术方案进行清楚、完整地描述。显然，所描述的实施例是本发明的一部分实施例，而不是全部的实施例。基于所描述的本发明的实施例，本领域普通技术人员所获得的所有其他实施例，都属于本发明保护的范围。

请参阅图1，图1为本发明实施例一的指标实时分析系统的结构示意图，该指标实时分析系统10包括：日志接入模块11、实时计算引擎模块12和存储模块13；

所述日志接入模块11，用于采集并解析原始日志数据，得到解析后的日志数据作为流数据，并将所述流数据写入所述存储模块13；

所述实时计算引擎模块12，用于从所述存储模块13获取所述流数据，对所述流数据进行计算，得到指标数据，并将所述指标数据写入所述存储模块13。

所述实时计算引擎模块12，与所述存储模块13连接，进一步的与所述日志接入模块11连接。

本发明实施例的指标实时分析系统，通过日志接入模块对原始日志进行解析，并将解析后的日志数据写入存储模块，实时计算引擎模块从存储模块中获取解析后的日志数据，计算得出指标数据，将计算好的指标数据，输出到存储模块中，能够加快数据处理的速度，保证指标计算的实时性，做到即时输出。

上述实施例中，解析后的日志数据可以为结构化日志数据，利于提升处理大量日志数据的速度。

上述实施例中，指标实时分析系统可以为统一日志平台，通过采集日志，实时计算指标，以进行实时监控。

优选的，所述日志接入模块采用至少一个服务器端数据采集组件(logstash组件)采集并解析所述原始日志数据。

上述实施例中，所述存储模块可以为分布式消息管道，优选的，所述存储模块采用分布式消息管道Kafka集群进行存储。

上述实施例中，实时计算引擎模块包括至少一个实时计算引擎，可以根据需求进行灵活定制，更为方便。

在本发明的一些优选实施例中，所述存储模块存储有日志数据对应的多个日志主题；

所述日志接入模块，用于根据所述流数据的日志属性，确定所述流数据对应的日志主题，将所述流数据存储至对应的日志主题中；

或者，

所述存储模块存储有日志数据对应的一个日志主题；

所述日志接入模块，用于根据所述流数据的日志属性，为所述流数据打上对应的标签，并将打上标签后的所述流数据存储至所述日志主题中。

也就是说，可以根据流数据的日志属性，例如：日志级别(info、error等)，建立索引，将流数据存储在不同的日志主题(Topic)下，以减轻存储的压力。而且，采用分布式消息管道Kafka集群作为存储模块，由于Kafka仅是消息管道，是否将日志数据存储在不同的日志主题中对于Kafka没有实质影响，即，既不会影响将流数据写入存储模块的速度，也不会影响实时计算引擎模块从存储模块读取流数据的速度，从而保证了指标计算的实时性。

也可以只将日志数据存在一个日志主题下，例如，将所有流数据存在Topic A下，为流数据打上对应的标签(如：用字段type表示流数据的属性，type字段可以是info、error等，为不同属性的流数据打上不同的标签)。在计算指标时，可对不同标签的流数据进行过滤。

在本发明的一些优选实施例中，所述实时计算引擎模块，用于根据指标的配置信息，生成与所述指标对应的实时计算任务，采用所述实时计算任务对所述流数据进行计算，得到所述指标数据，并将所述指标数据写入所述存储模块。

其中，所述指标可以为用户新建的指标或用户更新的指标，可根据实际需求配置指标，更加灵活方便。实时计算引擎模块中可运行一个或同时运行多个实时计算任务，一个实时计算任务可计算一个或多个指标。

本发明实施例中，多个日志Topic(例如：Topic A、Topic B、Topic C)可同时通过一个实时计算任务来计算指标，一个实时计算任务可以计算对应的一个或多个指标。而且，针对同一条流数据，可采用不同的过滤语句，执行相同的流任务来计算不同的指标。

在本发明的一些优选实施例中，所述存储模块存储有指标数据对应的多个指标主题；

所述实时计算引擎模块，还用于根据所述指标数据的属性，确定所述指标数据对应的指标主题，将所述指标数据存储至对应指标主题中。

也就是说，可以将计算好的指标数据存储为分主题的指标，即存储为指标Topic。

例如：按业务规则将指标数据分为多个指标主题，例如分为：机器的指标、路由器的指标、网关的指标等。

本发明实施例中，可以将指标数据双写(即存储同时备份)至第一数据库，便于用户对指标数据进行查询、分析及展示。

优选的，所述存储模块包括至少两个第一数据库；

所述存储模块，还用于将所述指标数据同时存储至所述至少两个第一数据库中。

也就是说，可以将指标数据同时存储至所述至少两个第一数据库中，能更好地保护数据安全，如：其中一个第一数据库发生数据丢失等情况，能通过存储于另一个数据库中的备份数据进行数据恢复，确保指标实时分析系统的持续稳健运行。

当然，也可以将指标数据存储于一个第一数据库的一文件夹中，备份于该第一数据库的另一文件夹中，本发明不作限定。

第一数据库可以为时间序列数据库Prometheus或基于HBase的分布式、可伸缩的时间序列数据库OpenTSDB(Open time series data base)。

在本发明的一些优选实施例中，所述存储模块，用于将第一预设时间之前的指标数据存储至第二数据库，将第一预设时间之后的指标数据存储于第一数据库。

具体而言，可以将历史的指标数据存储在第二数据库，将实时的指标数据存储于第一数据库。从而，可以在第二数据库查询离线的历史指标数据，在第一数据库查询实时的指标数据。

举例来说，可以将当天之前的指标数据存储至第二数据库，当天的指标数据(即实时的指标数据)存储至第一数据库。

优选的，所述第一数据库为Prometheus，所述第二数据库为OpenTSDB。

在本发明的一些优选实施例中，所述指标实时分析系统还包括：

查询模块，还用于根据用户的输入，产生指标查询命令；

展示模块，用于根据所述指标查询命令，查询所述存储模块中存储的所述指标数据，并展示查询到的所述指标数据。

具体而言，用户可以在用户界面(User Interface，简称UI)输入要查询的指标数据，查询模块产生相应的指标查询命令，展示模块从存储模块中查询并将查询到的指标数据进行展示，可以以数值或图表方式进行展示。

在本发明的一些优选实施例中，所述指标实时分析系统还包括：

配置中心，与所述实时计算引擎模块连接，用于存储所述指标的配置信息，所述指标的配置信息包括以下至少之一：所述指标的名称、所述指标的状态、所述指标的描述、所述指标的类型、所述指标的指标值类型、所述指标的标签和所述指标的计算定义；

所述实时计算引擎模块，还用于获取所述指标的配置信息，对所述指标的配置信息进行解析，根据解析结果，生成所述实时计算任务，并运行所述实时计算任务，所述实时计算引擎模块中运行有至少一个实时计算任务。

实时计算引擎模块中运行有一个或多个实时计算任务，能够提高资源的利用率，快速处理海量数据，加快计算速度。

优选的，所述配置中心采用Zookeeper组件实现。

优选的，指标实时分析系统还包括：指标配置模块，用于显示新建或更新所述指标的用户界面，并接收用户的新建或更新所述指标的操作，根据所述操作，新建或更新所述指标的配置信息，并存储至所述配置中心。

也就是说，用户可以在用户界面配置指标(即新建或更新指标)。以图2为示例，图2是本发明实施例的配置指标界面的显示示意图。在该配置指标界面(图2中具体为修改指标界面)20中，用户可以输入以下至少之一指标的配置信息：指标的名称、指标的状态、指标的描述、指标的类型、指标的指标值类型、指标的标签和指标的计算定义。

其中，指标的计算定义为指标的计算逻辑，可以采用SQL语言作为DSL(可编程式的计算机语言，domain specific language)，进行计算逻辑的定义，即指标的计算定义可采用SQL语言描述，SQL语言较为通用，能够降低用户的学习成本和使用门槛。用户在指标的计算定义处进行输入，可以新建指标，在指标的计算定义处进行修改，可以更新指标。

配置中心对用户在配置指标界面20输入的指标的配置信息进行存储，实时计算引擎模块获取指标的配置信息，生成实时计算任务，进行指标的实时计算。

本发明实施例中，实时计算引擎模块可以包括一个或多个实时计算引擎，例如：Flink、Spark和Storm至少之一，优选Flink引擎。将实时计算引擎模块的编辑调度器作为高层的处理引擎，Flink引擎作为底层的流处理框架，编辑调度器可用于告知Flink引擎要计算什么指标，Flink引擎将指标转化为实时计算任务，并进行执行，编辑调度器和Flink引擎两者配合使用，对流数据进行处理，可以做到毫秒级输出，实时性非常高，而且，利用分布式、高性能、高可用、高可靠的Flink引擎作为流处理框架，能够支持高并发、高吞吐、低延时的应用场景，也更为稳定。

本发明的实时计算引擎模块还支持管道(PIPELINE)，可以将前一级SQL的输出作为后一级SQL的输入，从而，可以对于单条SQL不方便处理的复杂计算逻辑提供支持。举例说明：

T1：

select tr.city,tr.serverity,MAX(tr.level)as maxlv,COUNT(tr.id)asrescnt from Tranlog as tr where tr.level>4000 and tr.accountId in('acct001','acct002’)group by tr.city,tr.serverity"

T2：

select MAX(rescnt)as maxpergrp,maxlv from T1

在交易日志中以城市和严重性(serverity)为分组策略，筛选出满足level大于4000且accountId为acct001或者acct002条件的城市、serverity、最大等级和个数。然后，从T1表中选出个数的最大值和最大等级。

在本发明的一些优选实施例中，所述实时计算引擎模块，用于针对每一所述实时计算任务，生成用于监听指标变化的注册命令，并提交至所述配置中心；

所述配置中心，用于当新建指标或更新指标时，生成携带所述新建指标或更新指标的配置信息的通知信息，向所述实时计算引擎模块中的每一所述实时计算任务广播所述通知信息。

具体而言，请参阅图3，图3是本发明实施例的配置中心与实时计算引擎模块的数据交互的流程示意图。实时计算引擎模块31包括监听子模块311、解析器312、缓存313和处理子模块314。

数据交互的具体流程如下：

3.1、监听子模块311用于针对每一实时计算任务，向配置中心32上注册监听指标变化的注册命令。

该注册命令用于监听指标是否新建或更新。通过监听配置中心32相应的配置节点，能够实现在不停止流计算的情况下，及时获取最新的指标(新建的指标或更新后的指标)，指标发生变化时，可以及时进行热更新，而无需停止或重启实时计算引擎，计算不间断，进一步提升了指标计算的实时性和准确性。

3.2、用户在用户界面33上新建指标或更新指标，新建的指标或更新的指标被存储至配置中心32中。

3.3、配置中心32生成携带新建指标或更新指标的配置信息的通知信息，通过监听子模块311向每一实时计算任务广播通知信息。

也就是说，有新建指标或更新指标时，配置中心32就生成通知信息，向实时计算引擎模块31中的每一计算实例广播对应的通知信号。

3.4、解析器312对新建的指标或更新的指标进行解析，将解析结果存储至缓存313。

该解析器312可以为内嵌的SQL解析器，具体可以为PARSER组件。

3.5、处理子模块315从缓存313中获取解析结果，运行实时计算任务，计算指标，将计算好的指标数据存储至缓存313。

具体而言，对每一条流数据，依条件进行过滤筛选，再依据指标的定义，对满足条件的流数据进行计算。时间范围变换时，指标的计算逻辑也可以被更新。例如，可基于时间窗口统计某级别/类型日志的数量：select count(*)from opm where level＝'Error'orlevel＝'Warning'group by index_date，基于日志中的时间、日志等级对流数据进行过滤筛选，对满足条件的日志数据进行统计计算，即进行计数。从而，针对每一条流数据进行指标的计算时，能够自动匹配满足的指标，这样新建的指标或更新的指标也会被实时计算引擎模块自动处理。

在本发明的一些优选实施例中，所述实时计算引擎模块，用于解析所述指标的计算定义，得到解析结果，所述解析结果中包括至少一个函数，其中，每一函数对应至少一个算子，将所述至少一个函数转换为至少一个算子，根据所述至少一个算子生成所述实时计算任务，所述实时计算任务包括：算子和算子之间的连接关系以及各个算子的配置。

其中，算子表示完成每种计算逻辑的计算抽象及上下游关联算子的数据传输。可以通过实时计算引擎模块中的解析器解析所述指标的计算定义，得到至少一个函数。

优选的，所述指标的计算定义包括以下函数中的至少之一：SELECT、FROM、JOIN、WHERE、GROUP BY、HAVING、ORDER BY；

所述算子包括以下至少之一：排序算子、表名匹配算子、关联表处理算子、WHERE条件匹配算子、聚合函数检测算子、聚合键生成算子、窗口检测算子、窗口计算算子、聚合计算算子、HAVING处理算子和字段选取算子。

具体而言，可以对用户输入的指标的计算定义进行解析，得到至少一个函数，将每一函数转换成至少一个算子，例如：SELECT函数用于选择具体用到的属性字段，对应字段选取算子；FROM函数对应表名匹配算子；JOIN函数对应关联表处理算子；WHERE函数对应WHERE条件匹配算子；GROUP BY函数对应聚合键生成算子、聚合计算算子；HAVING函数对应HAVING处理算子；ORDER BY函数对应排序算子(例如：基于交易时间排序)。

以图4为示例，图4是本发明实施例的算子实时计算的流程示意图。具体流程如下：

4.1、当一条流数据进入实时处理引擎模块，首先进入表名匹配算子，指标实时分析系统会遍历所有缓存的解析结果，例如SQL解析结果，根据流数据的表名，匹配需要被处理的指标的计算定义中的SQL语句。

其中，表名是数据所对应的消息主题名称，表名可以灵活定义，可以把日志的来源系统作为表名(日志的来源也可以与别的字段关联)。每一条流数据对应至少一个匹配的SQL语句。

4.2、关联表处理算子：根据预设关联条件，从静态数据源中查询与所述预设关联条件对应的关联数据，将关联数据添加至流数据中。

关联表处理算子用于流数据和静态数据源(例如：线下数据库)的关联，例如：在基于用户日志的行为分析场景中，往往要获取线下数据库例如关系型数据库中保存的用户信息等，可通过SQL子查询或者JOIN的方式将线上线下的数据无缝关联，比如：流数据中的交易数据有用户唯一标识(USER_ID)，但是还需要知道用户的姓名、年龄、信用级别等信息做进一步分析，但是这些信息存在线下数据库表里，可通过SQL语句的方式把这些信息从数据库里提取出来添加到流数据里，SQL语句如下：SELECT TRN.TRAN_ID,U.USER_CREDIT FROMTRANSACTION AS TRN JOIN USER_INFO AS U ON TRN.USER_ID＝USER_INFO.USER_ID.其中TRANSACTION是流式交易日志数据(包含USER_ID)，而USER_INFO是关系型数据库的表(包含用户的姓名、年龄、信用级别USER_CREDIT等信息)。该SQL语句的含义是：关联交易表(TRANSACTION)和用户信息表(USER_INFO)，当两个表的用户ID一致时，获取交易ID(在交易表中)和信用等级(在用户信息表中)。

从而，指标实时分析系统中可以对数据源集中管理，可根据4.1中解析得到的表名，和在预先保存的数据源中查找到的对应的数据类型进行关联。这样复杂的线上线下数据关联处理对用户完全透明，用户可以用统一的SQL语法来描述和处理，能进一步提升用户体验。

4.3、WHERE条件匹配算子：将流数据与第一预设条件进行匹配，输出匹配成功的流数据。

如果指标的计算定义存在WHERE函数，那么，指标实时分析系统会把流数据和WHERE条件做匹配，比如，采用SQL语言输入的指标的计算定义中，WHERE函数为WHEREUSERID>100，则，指标实时分析系统会检查流数据中是否存在USERID属性并且USERID的值>100，匹配成功的流数据才会传递给下一个算子。

4.4、聚合函数检测算子：标记流数据，以区分流数据对应的SELECT函数是否存在聚合函数；若SELECT函数存在聚合函数，将流数据输出至聚合键生成算子，否则，将流数据输出至字段选取算子。

基于实时计算引擎(如：Flink)本身的分流(SplitStream)特性和输出选择器(OutputSelector)特性，对流数据进行属性标记，检测流数据对应的SQL中SELECT函数里面是否有聚合函数，比如：COUNT/SUM等，如果有，流数据就进入聚合处理的算子，否则就进入字段选取算子。

4.5、聚合键生成算子：根据流数据对应的GROUP BY函数中的分组字段，查询流数据中的对应属性，根据对应属性形成键值，将相同键值的流数据输出至同一个下游算子。

如果指标的计算定义中存在GROUP BY函数，根据该GROUP BY函数中的分组字段，查询流数据中的对应属性形成键值，对数据进行分组，相同键值的为一组，进入同一个下游算子，确保聚合计算结果的正确性。该分组过程可对应Flink引擎的KEY BY算子。

4.6、窗口检测算子：检测流数据对应的指标的计算定义是否存在窗口函数；若存在窗口函数，将流数据输出至窗口计算算子。

流式计算的一大特性就是窗口计算，用于根据时间、数量、会话划分出一个一个有范围的窗口，计算在一定范围例如时间范围内的数据，可通过函数的方式支持窗口计算，比如：SELECT COUNT(TRAN_ID)WHERE WITHIN_WINDOW(‘TUMBLE’,10,‘MIN’)，即计算每10分钟的交易数量，算子如果检测到WHERE函数里有窗口函数，就把数据转到窗口计算算子，否则直接转到聚合计算算子。

4.7、窗口计算算子：将窗口内的流数据缓存至窗口结束后，将窗口内的流数据输出至聚合计算算子。

即，利用窗口将无限的流数据划分成一段一段的数据集，例如对于时间窗口在窗口内的流数据会被缓存至窗口结束时间到后触发计算，批量地发给聚合计算算子，从而对一批数据进行计算。

4.8、聚合计算算子：计算所述聚合函数的值，可以根据所述窗口内的所述流数据进行计算，也可以在没有窗口函数的情况下进行聚合计算；将所述聚合函数的值与所述流数据进行合并，所述聚合函数的值对应合并后的流数据中的一个属性，即衍生出一个属性。

可以针对利用窗口输出的流数据，计算指标的计算定义中存在的聚合函数的值。例如：SELECT USER_ID,MAX(AGE)FROM USER，指标的计算定义中有聚合函数，统计用户的最大年龄。SELECT USER_ID FROM USER，没有任何聚合函数，

按照聚合的键值和时间窗口，可以将计算结果保存在FLINK的内部状态中，再和当前的流数据进行合并，合并的结果保存为流数据属性。例如：原来的一条流数据包括A字段、B字段和C字段，计算聚合函数的值，将计算好的值作为D字段(衍生出一个属性)，加入到原来的一条数据中。

4.9、字段选取算子：根据流数据和SELECT函数中的选择字段，生成并输出新的流数据，每一SELECT函数中的选择字段对应新的流数据的一个属性。

举例来说，流数据可以为包含多列的数据，每一列对应一个字段即属性(例如：用户ID、交易ID、信用等级、交易时间等)，可以通过SELECT函数中的字段，对流数据进行筛选，筛选出包含部分属性的流数据，即为新的流数据。

至此，处理完成，新的流数据即计算出的指标数据从实时计算引擎模块输出至存储模块。

在本发明的另一些实施例中，若指标的计算定义中包括HAVING函数，流数据输入至聚合计算算子之前，可先输入至HAVING处理算子，经HAVING处理算子处理后再输入至聚合计算算子。

其中，HAVING处理算子用于将流数据与第二预设条件进行匹配，针对匹配成功的所述流数据，计算所述聚合函数的值。

也就是说，和WHERE条件匹配算子类似，HAVING处理算子也是对流数据进行过滤匹配，对匹配成功的数据进行聚合计算。若指标的计算定义中，没有HAVING函数，则在构建时，不需要对应的处理算子，即HAVING处理算子。

在本发明的一些优选实施例中，指标实时分析系统还包括：

函数自定义模块，用于获取自定义的函数，将自定义的函数注册至SQL解析的上下文中。

具体而言，函数自定义模块可以采用SQL语言描述指标的计算定义。当然，也可以采用别的语言描述，例如：函数自定义模块采用JAVA语言描述指标的计算定义，并通过配置的方式将JAVA类名/函数名注册到SQL解析的上下文中，这样，SQL处理的时候就可以无缝调用自定义的JAVA函数。从而，能够支持更多的函数和更复杂的功能，对复杂的计算逻辑进行处理。而现有技术中，指标的计算模式比较固定，仅提供一些固定模式的计算。由于指标不能灵活变化，如果业务上有其他需求，每需要新建或更新一个指标时，都需要单独开发一个程序，成本比较高。

在大数据系统中，数据倾斜是一个比较常见的问题。数据倾斜是指由于数据分布的问题导致不同的数据分布到不同处理节点的数量不均匀，有的处理节点工作量饱和甚至超出，而另一些节点则处于空闲状态，从而导致系统整体的吞吐量下降。

因此，在本发明的一些优选实施例中，所述实时计算任务中的算子包括多个实例，至少两个所述实例运行在不同的处理节点上；

所述实时计算引擎模块，用于从预设时间段内的所述流数据中抽取一数据样本，获取所述数据样本在所述不同处理节点上的数据分布比例；根据所述数据样本中的数据分布比例，分配之后时间段的在不同处理节点上的数据分布。

即，通过抽样的方式获得一个较小的样本，根据样本中在不同处理节点上的数据分布比例，预估一段时间内分布在不同处理节点上的数据的比例，进而分配之后时间段的在不同处理节点上的数据分布。

抽样的方式多种多样，可以按时间抽取正在处理的流数据，例如：每分钟抽取1条流数据，抽取一定时间后进行预估；也可以按数据抽取，例如：每100条流数据抽取1条，抽取一定数量后进行预估。

其中，一个算子可以在不同的处理节点上运行，一个处理节点可以计算多个算子的实例。一个物理机器(如：主机或服务器)可包括多个处理节点，或者，一个处理节点可以为一个物理机器。

可选的，所述实时计算引擎模块，用于获取所述数据样本中的流数据对应的键值，若一键值对应的流数据的数量大于或等于第一预设数量阈值，在所述一键值上增加一随机标识，以将所述一键值对应的流数据分配至不同处理节点；去除经所述不同处理节点处理后的流数据中的所述随机标识，合并去除所述随机标识后的流数据。

具体而言，指标的计算定义包括GROUP BY函数，根据流数据对应的键值将流数据分组，例如：将相同键值对应的流数据分为一组，每一键值与一组流数据唯一对应。若数据样本中某个键值的流数据的数量较多，则在之后时间段，处理该键值对应的流数据时，在该键值上增加随机标识(如：在该键值前加随机前缀)，这样，该键值对应的一组流数据会被打散至不同处理节点处理，处理完之后，再将随机标识去除，在下一个算子将流数据进行合并。

在数据关联过程中，为降低系统资源消耗，可选的，所述实时计算引擎模块，用于将流数据与静态数据进行关联时，获取所述流数据对应的预设键值的数量和所述静态数据对应的预设键值的数量，比较所述流数据对应的预设键值的数量与所述静态数据对应的预设键值的数量；

若所述流数据对应的预设键值的数量大于所述静态数据对应的预设键值的数量，向所述流数据广播所述静态数据对应的预设键值；

若所述流数据对应的预设键值的数量小于或等于所述静态数据对应的预设键值的数量，向所述静态数据广播所述流数据对应的预设键值；关联所述流数据与所述静态数据。

对于流数据和静态数据(例如：线下数据库)的关联，具体而言，指标的计算定义包括JOIN函数，例如：A JOIN B ON A.KEY1＝B.KEY2，A为流数据，例如A是交易流水，包含客户卡号和交易信息等，B为静态数据，例如B是线下数据库表，具体B为客户的基本信息表，包含客户账号、客户卡号和客户手机号等。预设键值为客户卡号。A JOIN B ON A.KEY1＝B.KEY2表示通过客户卡号进行A和B的关联，可得到新的C数据包含客户卡号、交易信息及客户手机号等，如果A中预设键值(KEY1，例如客户卡号)的数量非常多，而B中预设键值(KEY2，例如客户卡号)的数量少，则将B中的预设键值(客户卡号)广播(即传递)到A，通过预设键值关联A和B，使得A中的数据增加客户手机号和/或客户账号等，完成流数据A和静态数据B的关联，反之，如果A中预设键值的数量小于或等于B中的预设键值数量，则将A中的预设键值广播到B，通过预设键值关联A和B，使得A中的数据增加客户手机号和/或客户账号等，完成流数据A和静态数据B的关联，从而，可降低计算资源的消耗，进一步提升计算的速度。

在本发明的一具体应用场景中，请参阅图5，图5是本发明的一具体应用场景的指标实时分析系统的结构示意图。该指标实时分析系统50包括：日志接入模块51、实时计算引擎模块52和存储模块。其中，实时计算引擎模块52采用Flink引擎作为流处理框架；存储模块包括KAFKA集群531、Prometheus集群532和OpenTSDB集群533。日志接入模块51采用至少一个logstash组件511采集并解析原始日志数据，得到解析后的日志数据。将解析后的日志数据作为流数据，发送至KAFKA集群531中进行存储，存储为分主题的流数据。实时计算引擎模块52中运行至少一个实时计算任务，针对每一实时计算任务，从KAFKA集群531中获取流数据，根据指标的计算定义进行计算，得到指标数据，将指标数据输出至KAFKA集群531中进行存储，存储为分主题的指标数据。指标数据可以存储并备份到Prometheus集群532和OpenTSDB集群533。其中，实时指标数据(例如：当天的指标数据)优选存储在Prometheus集群532，历史指标数据(例如：当天之前的指标数据)存储在OpenTSDB集群533，可以在Prometheus集群532在线查询实时指标数据，在OpenTSDB集群533查询离线的历史指标数据。

在其他的具体应用场景中，指标数据可以只存储并备份到Prometheus集群，或者，只存储并备份到OpenTSDB集群，未在图5中示出，本发明不作限定。

本发明的具体应用场景的指标实时分析系统，通过实时计算引擎模块从KAFKA中获取解析后的日志数据，按照预先定义好的计算逻辑，对流数据进行计算，得出计算好的指标数据，并输出到KAFKA中。由于是实时的流式数据，可以做到即时输出(可以做到毫秒级的计算)，实时性非常高。

基于同样的发明构思，本发明还提供一种指标实时计算方法。请参阅图6，图6是本发明实施例二的指标实时计算方法的流程示意图，该指标实时计算方法包括：

步骤61：采集并解析原始日志数据，得到解析后的日志数据作为流数据，并存储所述流数据；

步骤62：获取所述流数据，对所述流数据进行计算，得到指标数据，并存储所述指标数据。

本发明实施例的指标实时计算方法，能够加快数据处理的速度，保证指标计算的实时性，做到即时输出。

优选的，所述存储所述流数据的步骤包括：

根据所述流数据的日志属性，确定所述流数据对应的日志主题，将所述流数据存储至对应的日志主题中，所述日志主题的个数为多个；

或者，

根据所述流数据的日志属性，为所述流数据打上对应的标签，并将打上标签后的所述流数据存储至一个日志主题中。

优选的，所述存储所述指标数据的步骤包括：

根据所述指标数据的属性，确定所述指标数据对应的指标主题，将所述指标数据存储至对应指标主题中。

优选的，所述存储所述指标数据的步骤包括：

将所述指标数据同时存储至至少两个第一数据库中。

优选的，所述存储所述指标数据的步骤包括：

将第一预设时间之前的指标数据存储至第二数据库，将第一预设时间之后的指标数据存储于第一数据库。

优选的，所述第一数据库为Prometheus，所述第二数据库为OpenTSDB。

优选的，所述指标实时计算方法还包括：

根据用户的输入，产生指标查询命令；

根据所述指标查询命令，查询所述指标数据，并展示查询到的所述指标数据。

优选的，所述对所述流数据进行计算，得到指标数据的步骤包括：

根据指标的配置信息，生成与所述指标对应的实时计算任务，采用所述实时计算任务对所述流数据进行计算，得到所述指标数据。

优选的，所述指标实时计算方法还包括：

存储所述指标的配置信息，所述指标的配置信息包括以下至少之一：所述指标的名称、所述指标的状态、所述指标的描述、所述指标的类型、所述指标的指标值类型、所述指标的标签和所述指标的计算定义；

获取所述指标的配置信息，对所述指标的配置信息进行解析，根据解析结果，生成所述实时计算任务，并运行所述实时计算任务。

优选的，所述指标实时计算方法还包括：

显示新建或更新所述指标的用户界面，并接收用户的新建或更新所述指标的操作，根据所述操作，新建或更新所述指标的配置信息，并存储新建或更新的指标的配置信息。

优选的，所述指标实时计算方法还包括：

针对每一所述实时计算任务，生成用于监听指标变化的注册命令；

当新建指标或更新指标时，生成携带所述新建指标或更新指标的配置信息的通知信息，向每一所述实时计算任务广播所述通知信息。

优选的，所述指标的配置信息包括所述指标的计算定义，所述指标实时计算方法还包括：

解析所述指标的计算定义，得到解析结果，所述解析结果中包括至少一个函数，其中，每一函数对应至少一个算子，将所述至少一个函数转换为至少一个算子，根据所述至少一个算子生成所述实时计算任务，所述实时计算任务包括：算子和算子之间的连接关系以及各个算子的配置。

优选的，所述指标的计算定义采用SQL语言描述。

优选的，所述指标的计算定义包括以下函数中的至少之一：SELECT、FROM、JOIN、WHERE、GROUP BY、HAVING、ORDER BY；

所述算子包括以下至少之一：排序算子、表名匹配算子、关联表处理算子、WHERE条件匹配算子、聚合函数检测算子、聚合键生成算子、窗口检测算子、窗口计算算子、聚合计算算子、HAVING处理算子和字段选取算子。

优选的，所述指标实时计算方法还包括：

获取自定义的函数，将自定义的函数注册至SQL解析的上下文中。

优选的，所述实时计算任务中的算子包括多个实例，至少两个所述实例运行在不同处理节点上；

所述指标实时计算方法还包括：从预设时间段内的所述流数据中抽取一数据样本，获取所述数据样本在所述不同处理节点上的数据分布比例；根据所述数据样本中的数据分布比例，分配之后时间段的在不同处理节点上的数据分布。

优选的，所述指标实时计算方法还包括：

获取所述数据样本中的流数据对应的键值，若一键值对应的流数据的数量大于或等于第一预设数量阈值，在所述一键值上增加一随机标识，以将所述一键值对应的流数据分配至不同处理节点；

去除经所述不同处理节点处理后的流数据中的所述随机标识，合并去除所述随机标识后的流数据。

优选的，所述指标实时计算方法还包括：

将流数据与静态数据进行关联时，获取所述流数据对应的预设键值的数量和所述静态数据对应的预设键值的数量；

比较所述流数据对应的预设键值的数量与所述静态数据对应的预设键值的数量；

若所述流数据对应的预设键值的数量大于所述静态数据对应的预设键值的数量，向所述流数据广播所述静态数据对应的预设键值；

若所述流数据对应的预设键值的数量小于或等于所述静态数据对应的预设键值的数量，向所述静态数据广播所述流数据对应的预设键值；

关联所述流数据与所述静态数据。

请参阅图7，图7是本发明实施例三的指标实时分析系统的结构示意图，该指标实时分析系统70包括：存储器71、处理器72及存储在所述存储器71上并可在所述处理器72上运行的计算机程序；所述处理器72执行所述计算机程序时实现如下步骤：

采集并解析原始日志数据，得到解析后的日志数据作为流数据，并存储所述流数据；

获取所述流数据，对所述流数据进行计算，得到指标数据，并存储所述指标数据。

本发明实施例的指标实时分析系统，能够加快数据处理的速度，保证指标计算的实时性，做到即时输出。

优选的，所述处理器72执行所述计算机程序时实现如下步骤：

所述存储所述流数据的步骤包括：

根据所述流数据的日志属性，确定所述流数据对应的日志主题，将所述流数据存储至对应的日志主题中，所述日志主题的个数为多个；

或者，

根据所述流数据的日志属性，为所述流数据打上对应的标签，并将打上标签后的所述流数据存储至一个日志主题中。

优选的，所述处理器72执行所述计算机程序时实现如下步骤：

所述存储所述指标数据的步骤包括：

根据所述指标数据的属性，确定所述指标数据对应的指标主题，将所述指标数据存储至对应指标主题中。

优选的，所述存储所述指标数据的步骤包括：

将所述指标数据同时存储至至少两个第一数据库中。

优选的，所述处理器72执行所述计算机程序时实现如下步骤：

所述存储所述指标数据的步骤包括：

将第一预设时间之前的指标数据存储至第二数据库，将第一预设时间之后的指标数据存储于第一数据库。

优选的，所述第一数据库为Prometheus，所述第二数据库为OpenTSDB。

优选的，所述处理器72执行所述计算机程序时实现如下步骤：

根据用户的输入，产生指标查询命令；

根据所述指标查询命令，查询所述指标数据，并展示查询到的所述指标数据。

优选的，所述处理器72执行所述计算机程序时实现如下步骤：

所述对所述流数据进行计算，得到指标数据的步骤包括：

根据指标的配置信息，生成与所述指标对应的实时计算任务，采用所述实时计算任务对所述流数据进行计算，得到所述指标数据。

优选的，所述处理器72执行所述计算机程序时实现如下步骤：

存储所述指标的配置信息，所述指标的配置信息包括以下至少之一：所述指标的名称、所述指标的状态、所述指标的描述、所述指标的类型、所述指标的指标值类型、所述指标的标签和所述指标的计算定义；

获取所述指标的配置信息，对所述指标的配置信息进行解析，根据解析结果，生成所述实时计算任务，并运行所述实时计算任务。

优选的，所述处理器72执行所述计算机程序时实现如下步骤：

显示新建或更新所述指标的用户界面，并接收用户的新建或更新所述指标的操作，根据所述操作，新建或更新所述指标的配置信息，并存储新建或更新的指标的配置信息。

优选的，所述处理器72执行所述计算机程序时实现如下步骤：

针对每一所述实时计算任务，生成用于监听指标变化的注册命令；

当新建指标或更新指标时，生成携带所述新建指标或更新指标的配置信息的通知信息，向每一所述实时计算任务广播所述通知信息。

优选的，所述指标的配置信息包括所述指标的计算定义，所述处理器72执行所述计算机程序时实现如下步骤：

解析所述指标的计算定义，得到解析结果，所述解析结果中包括至少一个函数，其中，每一函数对应至少一个算子，将所述至少一个函数转换为至少一个算子，根据所述至少一个算子生成所述实时计算任务，所述实时计算任务包括：算子和算子之间的连接关系以及各个算子的配置。

优选的，所述指标的计算定义采用SQL语言描述。

优选的，所述指标的计算定义包括以下函数中的至少之一：SELECT、FROM、JOIN、WHERE、GROUP BY、HAVING、ORDER BY；

所述算子包括以下至少之一：排序算子、表名匹配算子、关联表处理算子、WHERE条件匹配算子、聚合函数检测算子、聚合键生成算子、窗口检测算子、窗口计算算子、聚合计算算子、HAVING处理算子和字段选取算子。

优选的，所述处理器72执行所述计算机程序时实现如下步骤：

获取自定义的函数，将自定义的函数注册至SQL解析的上下文中。

优选的，所述实时计算任务中的算子包括多个实例，至少两个所述实例运行在不同处理节点上；

所述指标实时计算方法还包括：从预设时间段内的所述流数据中抽取一数据样本，获取所述数据样本在所述不同处理节点上的数据分布比例；根据所述数据样本中的数据分布比例，分配之后时间段的在不同处理节点上的数据分布。

优选的，所述处理器72执行所述计算机程序时实现如下步骤：

获取所述数据样本中的流数据对应的键值，若一键值对应的流数据的数量大于或等于第一预设数量阈值，在所述一键值上增加一随机标识，以将所述一键值对应的流数据分配至不同处理节点；

去除经所述不同处理节点处理后的流数据中的所述随机标识，合并去除所述随机标识后的流数据。

优选的，所述处理器72执行所述计算机程序时实现如下步骤：

将流数据与静态数据进行关联时，获取所述流数据对应的预设键值的数量和所述静态数据对应的预设键值的数量；

比较所述流数据对应的预设键值的数量与所述静态数据对应的预设键值的数量；

若所述流数据对应的预设键值的数量大于所述静态数据对应的预设键值的数量，向所述流数据广播所述静态数据对应的预设键值；

若所述流数据对应的预设键值的数量小于或等于所述静态数据对应的预设键值的数量，向所述静态数据广播所述流数据对应的预设键值；

关联所述流数据与所述静态数据。

本发明实施例四提供一种计算机可读存储介质，其上存储有计算机程序，该计算机程序被处理器执行时实现上述实施例二中指标实时计算方法中的步骤。具体工作过程与上述对应实施例二中的一致，故在此不再赘述，详细请参阅以上对应实施例中方法步骤的说明。

以上所述是本发明的优选实施方式，应当指出，对于本技术领域的普通技术人员来说，在不脱离本发明所述原理的前提下，还可以作出若干改进和润饰，这些改进和润饰也应视为本发明的保护范围。