数据聚合处理方法、计算节点、计算集群及存储介质

摘要

本发明公开了一种数据聚合处理方法、计算节点、计算集群及存储介质，该数据聚合处理方法应用于计算集群中的计算节点，该方法包括：获取初始值和所述计算节点的节点数据，其中，所述节点数据和所述初始值的存储方式为列式存储；基于列式遍历的方式，对所述初始值和所述节点数据进行处理，以生成节点处理数据；若所述计算节点为处理节点，则将所述节点处理数据发送至所述计算集群的聚合节点；若所述计算节点为所述聚合节点，则将各个所述节点处理数据聚合，得到集群聚合数据，以将所述集群聚合数据发送至客户端，实现了基于计算集群，对列式存储的数据，采用列式遍历的方式进行聚合，提高了数据聚合的效率。

说明书

技术领域

本发明涉及大数据处理技术领域，尤其涉及一种数据聚合处理方法、计算节点、计算集群及存储介质。

背景技术

随着互联网的飞速发展，数据量急剧增长，大数据分析已应用于各个领域。为了提高大数据处理的效率，通常需要基于分布式计算架构进行大数据处理。

然而，由于当前的分布式计算架构大多采用行式存储方式，如spark，从而其分布式算子，如map、reduce、aggregate等，均采用行式计算的方式，然而对于一些用户需求为需要针对其他存储形式的数据进行处理时，如需要计算其他存储形式的数据的最大值、平均值等，当所需处理的数据量较大时，采用行式计算方式的算子的计算开销较大，计算效率较低，无法满足需求。

发明内容

本发明的主要目的在于提供一种数据聚合处理方法、计算节点、计算集群及存储介质，针对基于列式存储方式的数据，提供了一种高速有效的聚合处理方法，提高了分布式存储的大数据聚合处理的性能。

为实现上述目的，第一方面，本发明实施例提供一种数据聚合处理方法，该方法应用于计算集群中的计算节点，该数据聚合处理方法包括：

获取初始值和所述计算节点的节点数据，其中，所述节点数据和所述初始值的存储方式为列式存储；基于列式遍历的方式，对所述初始值和所述节点数据进行处理，以生成节点处理数据；若所述计算节点为处理节点，则将所述节点处理数据发送至所述计算集群的聚合节点；若所述计算节点为所述聚合节点，则将各个所述节点处理数据聚合，得到集群聚合数据，以将所述集群聚合数据发送至客户端。

可选地，基于列式遍历的方式，对所述初始值和所述节点数据进行处理，以生成节点处理数据，包括：

按照列式遍历的方式，读取所述计算节点的所述节点数据和所述初始值的每列数据；基于预设逻辑算法，根据所述节点数据和所述初始值的每列数据，生成所述计算节点的所述节点处理数据。

可选地，按照列式遍历的方式，读取所述计算节点的所述节点数据和所述初始值的每列数据，包括：

根据所述计算节点的线程的数量，将所述节点数据和所述初始值按列进行划分，以得到多个节点子数据和多个子初始值，其中，所述多个节点子数据的数量以及所述多个子初始值的数量均与所述线程的数量一致；针对所述计算节点的每个线程，基于列式遍历的方式，读取所述线程对应的节点子数据和子初始值的每列数据。

相应的，基于预设逻辑算法，根据所述节点数据和所述初始值的每列数据，生成所述计算节点的所述节点处理数据，包括：

针对所述计算节点的每个线程，基于预设逻辑算法，根据所述线程的所述节点子数据和所述子初始值的每列数据，生成所述线程的线程处理数据；根据各个所述线程处理数据生成所述计算节点的所述节点处理数据。

可选地，在根据所述计算节点的线程的数量，将所述节点数据和所述初始值按列进行划分之前，所述方法还包括：

根据所述计算节点的所述节点数据和所述初始值，生成所述计算节点的各个线程共享的共享变量。

相应的，针对所述计算节点的每个线程，基于预设逻辑算法，根据所述线程的所述节点子数据和所述子初始值的每列数据，生成所述线程的线程处理数据，包括：

针对所述计算节点的每个需要调用共享变量的线程，基于预设逻辑算法，根据所述共享变量、所述线程的所述节点子数据和所述子初始值的每列数据，生成所述线程的线程处理数据；针对所述计算节点的每个不需要调用共享变量的线程，基于预设逻辑算法，根据所述线程的所述节点子数据和所述子初始值的每列数据，生成所述线程的线程处理数据。

可选地，将各个节点处理数据聚合，得到集群聚合数据，包括：

获取第一计算节点的节点处理数据；当存在至少一个第二计算节点的节点处理数据时，则将所述第一计算节点的节点处理数据和所述至少一个第二计算节点的节点处理数据聚合，得到中间聚合数据，其中，所述第一计算节点为所述计算集群中的任意一个所述计算节点，所述第二计算节点为所述计算集群中的所述第一计算节点不同的所述计算节点；将所述中间聚合数据放置于所述节点处理数据对应的队列中，并删除所述第一计算节点的节点数据和至少一个第二计算节点的节点处理数据；当各个所述计算节点的节点处理数据均被聚合，且所述节点处理数据对应的队列中的数据的数量为1时，聚合完毕，所述节点处理数据对应的队列中的数据为所述集群聚合数据。

可选地，所述计算集群包括多个服务器，每一服务器包括至少一个计算节点；在生成所述计算节点的所述节点处理数据之后，还包括：

若存在与所述计算节点属于同一服务器的其它计算节点，则经由所述服务器的预设计算节点，将所述服务器的各个计算节点的所述节点处理数据聚合，以得到服务器聚合数据，其中，所述预设计算节点为任意一个所述服务器的所述计算节点。

相应的，将各个所述节点处理数据聚合，得到集群聚合数据，包括：

经由所述聚合节点将所述计算集群的各个服务器的服务器聚合数据聚合，得到集群聚合数据。

可选地，在获取计算节点的节点数据之后，所述方法还包括：

将所述节点数据的存储方式转换为堆外内存的存储方式。

第二方面，本发明实施例还提供一种数据聚合处理装置，包括：

数据获取模块，用于获取初始值和计算节点的节点数据，其中，所述节点数据和所述初始值的存储方式为列式存储；数据处理模块，用于基于列式遍历的方式，对所述初始值和所述节点数据进行处理，以生成节点处理数据；数据发送模块，用于若所述计算节点为处理节点，则将所述节点处理数据发送至所述计算集群的聚合节点；数据聚合模块，用于若所述计算节点为所述聚合节点，则将各个所述节点处理数据聚合，得到集群聚合数据，以将所述集群聚合数据发送至客户端。

第三方面，本发明实施例还提供一种计算节点，所述计算节点包括：存储器、处理器及存储在所述存储器上并可在所述处理器上运行的数据聚合处理程序，所述数据聚合处理程序被所述处理器执行时实现如本发明第一方面对应的任意实施例提供的数据聚合处理方法的步骤。

第四方面，本发明实施例还提供一种计算集群，包括：本发明第三方面对应的实施例提供的多个计算节点。

第五方面，本发明实施例还提供一种计算机可读存储介质，所述计算机可读存储介质上存储有数据聚合处理程序，所述数据聚合处理程序被处理器执行时实现如本发明第一方面对应的任意实施例提供的数据聚合处理方法的步骤。

第六方面，本发明实施例还提供一种计算机程序产品，包括计算机程序/指令，该计算机程序/指令被处理器执行时实现如本发明第一方面对应的任意实施例提供的数据聚合处理方法的步骤。

本发明实施例提供的数据聚合处理方法、计算节点、计算集群及存储介质，针对分布式存储的节点数据，由计算集群的计算节点获取对应的节点数据，以及来自用户的客户端的初始值，其中，该初始值和节点数据均采用列式存储的方式进行存储，计算节点中的相关算子则采用列式遍历的方式对初始值和节点数据进行处理，从而得到节点处理数据，进而由计算集群中的聚合节点将各个节点处理数据聚合，得到最终的集群聚合数据，从而将该集群聚合数据发送至客户端，进行相应的业务处理，针对采用列式存储方式的数据，提供了基于列式遍历的聚合算法，针对按列进行聚合的用户需求，提供了高效的列式聚合算子，提高了数据处理速度和数据聚合的效率。

附图说明

图1是本发明实施例提供的数据聚合处理方法的一种应用场景图；

图2是本发明实施例提供的数据聚合处理方法的流程图；

图3是本发明图2所示实施例中列式存储的内存数据的示意图；

图4是本发明图2所示实施例中步骤S202的流程图；

图5是本发明另一个实施例提供的数据聚合处理方法的流程图；

图6是本发明图5所示实施例中步骤S407的流程图；

图7是本发明另一实施例提供的数据聚合处理方法的流程图；

图8为本发明一个实施例提供的数据聚合处理设备的结构示意图；

图9为本发明一个实施例提供的计算节点的结构示意图；

图10为本发明一个实施例提供的计算集群的结构示意图。

本发明目的的实现、功能特点及优点将结合实施例，参照附图做进一步说明。

具体实施方式

下面将参照附图更详细地描述本公开的示例性实施例。虽然附图中显示了本公开的示例性实施例，然而应当理解，可以以各种形式实现本公开而不应被这里阐述的实施例所限制。相反，提供这些实施例是为了能够更透彻地理解本公开，并且能够将本公开的范围完整的传达给本领域的技术人员。

下面对本发明实施例的应用场景进行解释：

图1是本发明实施例提供的数据聚合处理方法的一种应用场景图，随着互联网的不断发展，数据量急剧增长，大数据分析已广泛应用于各行各业，为了提高数据分析的效率，往往采用包括多个计算节点的计算集群进行大数据分析。计算集群110往往包括多个计算节点111；当用户端120存在业务需求，往往需要对分布式存储的业务数据130进行聚合；具体为由计算集群110的各个计算节点111加载其对应的部分业务数据，作为该计算节点111的节点数据，并基于该计算节点的聚合算子对节点数据进行处理，通常该聚合算子是通过相应的分布式计算架构设计的；进而整合各个计算节点111处理后的节点数据，得到计算集群的聚合数据140；将该聚合数据140发送至用户端120，以使用户端120基于该聚合数据140开展后续的业务处理。具体的业务可以是根据聚合数据140确定用于模型训练的训练数据，以根据训练数据进行模型训练，相应的业务数据130便为用于模型训练的原始数据。业务需求还可以是对某个企业的各个子公司的各个员工的年平均收入进行计算，业务数据130便可以是各个子公司的各个员工每个月的收入，聚合数据140便为各个子公司的各个员工的年平均收入，用户端120便可以根据各个子公司的各个员工的年平均收入，对各个子公司进行评级或者其他业务处理。当然，业务需求还可以是其他形式的需求，本申请对此不进行限制。

然而，现有的聚合算子，如spark中的map、reduce、aggregate等算子，由于其采用的分布式计算架构本身的限制，如spark中采用的RDD(Resilient Distributed DataSet，弹性分布式数据集)是基于行式存储的，从而导致聚合算子仅可以采用行式计算的方式进行，从而使得列式计算需求的业务处理效率较低，无法满足用户需求。

为了提高列式计算需求业务对应的数据聚合的效率，本发明实施例提供的数据聚合处理方法的主要构思为：将计算节点中的节点数据和初始值的存储方式设置为列式存储，并基于列式存储的节点数据和初始值，采用列式遍历的方式进行数据处理，进而聚合处理后的数据，得到用户所需的聚合数据，本发明提供的数据聚合处理方法，兼顾了列式处理业务的效率需求以及聚合算子与其对应的分布式架构的兼容性。

图2是本发明实施例提供的数据聚合处理方法的流程图，该数据聚合处理方法可以运行于计算集群中的计算节点上，可以是任意一个计算节点，如图2所示，该数据聚合处理方法包括以下步骤：

步骤S201，获取初始值和所述计算节点的节点数据。

其中，所述节点数据和所述初始值的存储方式为列式存储。

具体的，列式存储的存储方式中，数据每一列的数据类型是相同的，且数据地址是连续分布的，而不同列之间的数据地址可以是不连续的，数据类型也可以不同。

示例性的，图3是本发明图2所示实施例中列式存储的内存数据的示意图，结合图3可知，该内存数据为存储于计算节点内的数据，可以是上述初始值或者节点数据，该内存数据包括多列数据，第一列数据为整型数据(int)，而第二列数据为双精度浮点型数据(double)，最后一列数据则为单精度浮点型数据(float)，每一列数据对应的数据类型是相同的。

具体的，初始值zeroValue可以是由客户端广播或发送至计算集群的计算节点的，该客户端可以是计算集群中的一个节点，还可以是与计算集群通讯的其他计算节点。该初始值zeroValue的具体内容可以由用户自定义，还可以采用默认值。节点数据inputValue可以是从任意来源获取的业务数据的一部分，是用户需要进行聚合处理的业务数据的一部分，该业务数据inputValue为采用分布式存储的数据。该节点数据inputValue可以是用于进行模型训练的训练数据的一部分，如可以是用于训练决策树的训练数据的一部分，当然也可以是训练其他模型，如神经网络模型，的训练数据的一部分。

步骤S202，基于列式遍历的方式，对所述初始值和所述节点数据进行处理，以生成节点处理数据。

其中，列式遍历的方式即为按照数据的列进行数据遍历，从而获取每列的数据，进行数据处理，得到节点处理数据。

具体的，可以经由计算节点的聚合算子，基于列式遍历的方式，对初始值和节点数据进行处理，以生成节点处理数据。该聚合算子可以是基于分布式架构设计的分布式算子，具体可以是基于spark的改进的aggregate分布式算子。

进一步地，该改进的aggregate分布式算子为基于列式遍历方式进行数据遍历的aggregate分布式算子。

可选地，图4是本发明图2所示实施例中步骤S202的流程图，如图4所示，步骤S202包括以下步骤：

步骤S2021，按照列式遍历的方式，读取所述节点数据和所述初始值的每列数据。

具体的，可以是按照列式遍历的方式，在每次遍历时，一次性遍历初始值和节点数据的完整的一列或多列数据，将该完整的一列或多列数据存储于缓存中，当该完整的一列或多列数据调用完毕后，释放缓存，进而加载下一次遍历对应的完整的一列或多列数据至缓存中。采用遍历完整一列或多列数据的方式，有效避免了出现间断性访问不同列数据的情况。

由于计算节点的CPU(Central Processing Unit，中央处理器)的缓存机制为：在按照列式遍历的方式，遍历初始值和节点数据的一列数据时，会预先将该列数据加载得到缓存中，进而加快CPU读取该列数据的速度，当该列数据调用完毕后，该缓存便会释放，进而加载下一列数据至该缓存中，当出现间断访问不同列的数据时，便会导致频繁在不同列的数据之间跳转，从而使得缓存不断加载和释放数据，导致数据处理效率低下。

步骤S2022，基于预设逻辑算法，根据所述节点数据和所述初始值的每列数据，生成所述计算节点的所述节点处理数据。

其中，预设逻辑算法为用户根据其业务需求设计的逻辑算法。

具体的，预设逻辑算法可以包括针对每列数据的求和运算、最大值运算、最小值运算、方差运算、平均值运算等算法，当然还可以其他复杂的运算对应的算法，该预设逻辑算法的具体算法逻辑由用户自定义设计。

进一步地，还可以根据用户需求，先按照列式遍历的方式，读取所述计算节点的所述节点数据和所述初始值的每列数据，再按照行式遍历的方式，读取计算节点的所述节点数据和所述初始值的每行数据，进而基于预设逻辑算法，根据所述节点数据和所述初始值的每列数据，以及所述节点数据和所述初始值的每行数据，生成所述计算节点的所述节点处理数据。

步骤S203，若所述计算节点为处理节点，则将所述节点处理数据发送至所述计算集群的聚合节点。

其中，聚合节点是计算集群中用于将各个处理节点的节点处理数据聚合的计算节点，处理节点是计算集群中的除聚合节点之外的任意一个计算节点。

具体的，当计算节点基于预设逻辑算法，采用列式遍历的方式，将节点数据和初始值处理完毕后，生成该计算节点的节点数据，进而当该计算节点不是用于执行聚合操作的计算节点，即聚合节点时，则将所生成的节点数据发送至聚合节点。

步骤S204，若所述计算节点为所述聚合节点，则将各个所述节点处理数据聚合，得到集群聚合数据，以将所述集群聚合数据发送至客户端。

具体的，可以通过聚合节点的聚合算子的聚合模块，将各个所述节点处理数据聚合，得到集群聚合数据。

具体的，当计算节点为执行聚合操作的计算节点，即聚合节点时，则由该聚合节点获取各个计算节点，包括其自身，生成的节点处理数据，并将各个节点处理数据聚合，得到集合聚合数据，将该集合聚合数据发送至客户端，以使客户端根据该集合聚合数据进行相关业务处理。

需要说明的是，当计算节点为处理节点时，则无需进行步骤S204的相关操作，即当计算节点为处理节点时，其对应的数据聚合处理方法仅包括步骤S201～步骤S203。而当计算节点为聚合节点时，则无需进行步骤S203的相关操作即当计算节点为聚合节点时，其对应的数据聚合处理方法仅包括步骤S201、步骤S202和步骤S204。

在本实施例中，针对分布式存储的节点数据，由计算集群的计算节点获取对应的节点数据，以及来自用户的客户端的初始值，其中，该初始值和节点数据均采用列式存储的方式进行存储，计算节点中的相关算子则采用列式遍历的方式对初始值和节点数据进行处理，从而得到节点处理数据，进而由计算集群中的聚合节点将各个节点处理数据聚合，得到最终的集群聚合数据，从而将该集群聚合数据发送至客户端，进行相应的业务处理，针对采用列式存储方式的数据，提供了基于列式遍历的聚合算法，针对按列进行聚合的用户需求，提供了高效的列式聚合算子，提高了数据处理速度和数据聚合的效率。

图5是本发明另一个实施例提供的数据聚合处理方法的流程图，本实施例是在图2所示实施例的基础上，对步骤S202和步骤S204的进一步细化，以如图5所示，本实施例提供的数据聚合处理方法包括以下步骤：

步骤S401，获取初始值和所述计算节点的节点数据。

其中，所述节点数据和所述初始值的存储方式为列式存储。

可选地，在获取计算节点的节点数据之后，所述方法还包括：将所述节点数据的存储方式转换为堆外内存的存储方式。堆外内存的存储方式具有零拷贝的特点，可以提高数据读取和复用的效率。

具体的，在对分布式存储的业务数据进行聚合之前，需要为计算集群的每个计算节点分配相应的任务，即需要确定各个计算节点需要处理的节点数据，该节点数据的来源可以是任意的，可以是存储在计算节点的磁盘上或者其他位置处，还可以是存储于其他设备上。当计算节点获取该节点数据之后，为了提高数据处理的效率，需要将节点数据的存储方式转换为堆外内存的存储方式。当然，若节点数据的存储方式本身便是堆外内存的存储方式，则可以忽略本步骤。

同样的，对于初始值来说，在获取初始值之后，还可以将所述初始值的存储方式转换为堆外内存的存储方式。

步骤S402，根据所述计算节点的线程的数量，将所述节点数据和所述初始值按列进行划分，以得到多个节点子数据和多个子初始值。

其中，所述多个节点子数据的数量以及所述多个子初始值的数量均与所述线程的数量一致。

为了加快数据处理的速度，可以采用多线程运行的方式，对节点数据和初始值进行处理。计算节点的线程的数量可以由计算节点的空闲的各个CPU的核心的数量确定。具体的，用于节点数据和初始值处理的线程的数量可以是空闲的各个CPU的核心的数量与1的差。

具体的，假设线程的数量为n，则将初始值zeroValue和节点数据inputValue按列拆分为n份，从而得到n个节点子数据和n个子初始值，但不改变每个数据，即节点子数据和子初始值，对应的内存地址。

步骤S403，针对所述计算节点的每个线程，基于列式遍历的方式，读取所述线程对应的节点子数据和子初始值的每列数据。

具体的，计算节点的每个线程，均采用列式遍历的方式，对该线程对应的节点子数据和子初始值进行每列数据的读取。

步骤S404，针对所述计算节点的每个线程，基于预设逻辑算法，根据所述线程的所述节点子数据和所述子初始值的每列数据，生成所述线程的线程处理数据。

具体的，计算节点的每个线程，通过运行用户设计的预设逻辑算法，对该线程的节点子数据和子初始值的每列数据进行相应的处理，得到线程对应的线程处理数据。

可选地，在根据所述计算节点的线程的数量，将所述节点数据和所述初始值按列进行划分之前，还包括：根据所述计算节点的所述节点数据和所述初始值，生成所述计算节点的各个线程共享的共享变量。相应的，针对所述计算节点的每个线程，基于预设逻辑算法，根据所述线程的所述节点子数据和所述子初始值的每列数据，生成所述线程的线程处理数据，包括：针对所述计算节点的每个需要调用共享变量的线程，基于预设逻辑算法，根据所述共享变量、所述线程的所述节点子数据和所述子初始值的每列数据，生成所述线程的线程处理数据；针对所述计算节点的每个不需要调用共享变量的线程，基于预设逻辑算法，根据所述线程的所述节点子数据和所述子初始值的每列数据，生成所述线程的线程处理数据。

其中，共享变量为至少两个线程均需要调用的变量，该共享变量可以理解为预设逻辑算法中，至少两个线程，可以是每个线程，需要重复计算的部分。

示例性的，假设计算节点包括3个线程，根据预设逻辑算法，确定每个线程均需要得到节点数据与初始值的和，则该共享变量变为节点数据与初始值的和。当然，共享变量、初始值和节点数据之间的对应关系还可以是其他关系，具体由用户设计的预设逻辑算法确定。

步骤S405，根据各个所述线程处理数据生成所述计算节点的所述节点处理数据。

具体的，根据所述线程对应的节点数据的列的位置，整合或者合并各个线程处理数据，得到该计算节点的节点处理数据。

进一步地，该计算节点的节点处理数据可以存储于该计算节点对应的初始值的变量对应的地址处，即将计算节点处理得到的节点处理数据直接写到初始值中，从而避免生成新的中间变量存储该节点处理数据，减少所需的内存。

步骤S406，若所述计算节点为处理节点，则将所述节点处理数据发送至所述计算集群的聚合节点。

可选地，计算集群包括多个服务器，每个服务器包括至少一个计算节点，当服务器包括至少两个计算节点时，在生成所述计算节点的所述节点处理数据之后，还包括：若存在与所述计算节点属于同一服务器的其它计算节点，则经由所述服务器的预设计算节点，将所述服务器的各个计算节点的所述节点处理数据聚合，以得到服务器聚合数据，其中，所述预设计算节点为任意一个所述服务器的所述计算节点。相应的，后续的将各个所述节点处理数据聚合，得到集群聚合数据，包括：经由所述聚合节点将所述计算集群的各个服务器的服务器聚合数据聚合，得到集群聚合数据。

具体的，若服务器存在多个计算节点，如3个、5个或者其他数量，在需要以服务器为单位，将该服务器的各个计算节点的节点处理数据聚合，得到服务器聚合数据，进而由聚合节点聚合各个服务器的服务器聚合数据，得到集群聚合数据。这样设置的好处在于，可以通过多计算节点的服务器并行将该服务器的各个节点处理数据聚合，从而减少聚合节点所需聚合的数据量，提高数据聚合的效率。

步骤S407，若所述计算节点为所述聚合节点，将各个所述节点处理数据聚合，得到集群聚合数据，并将所述集群聚合数据发送至客户端。

图6是本发明图5所示实施例中步骤S407的流程图，如图6所示，步骤S407包括以下步骤：

步骤S4071，若所述计算节点为所述聚合节点，则获取第一计算节点的节点处理数据。

其中，所述第一计算节点为所述计算集群中的任意一个所述计算节点。

步骤S4072，当存在至少一个第二计算节点的节点处理数据时，则将所述第一计算节点的节点处理数据和所述至少一个第二计算节点的节点处理数据聚合，得到中间聚合数据。

其中，所述第二计算节点为所述计算集群中的所述第一计算节点不同的所述计算节点。

步骤S4073，将所述中间聚合数据放置于所述节点处理数据对应的队列中，并删除所述第一计算节点的节点数据和至少一个第二计算节点的节点处理数据。

步骤S4074，当各个所述计算节点的节点处理数据均被聚合，且所述节点处理数据对应的队列中的数据的数量为1时，聚合完毕，所述节点处理数据对应的队列中的数据为所述集群聚合数据，并将所述集群聚合数据发送至客户端。

具体的，可以通过聚合节点的聚合算子的聚合模块，执行步骤S4071～步骤S4072。

具体的，该聚合节点可以包括多个线程，可以由聚合节点的每个线程，获取不同的至少两个计算节点的节点处理数据和/或中间聚合数据，进行聚合，进而得到集群聚合数据。

具体的，对于聚合节点来说，由于各个处理节点的计算速度以及通信传输时间的差异，导致聚合节点接收到各个计算节点的节点处理数据的时间不一样，当接收到全部的处理节点的节点处理数据之后，再进行聚合，会造成时间浪费。因此，需要当其线程中存在两个或两个以上的节点处理数据，便基于一个线程对节点处理数据进行聚合操作，进而得到中间聚合数据，与此同时，而无需等到获取全部计算节点的节点数据之后，再进行聚合，减少了等待数据的时间，提高了数据聚合的效率。

进一步地，可以根据获取到节点处理数据的时间以及生成中间聚合数据的时间，确定各个节点处理数据和中间聚合数据在队列中的位置，并删除处理过的节点处理数据和中间聚合数据，从而当队列中的数据的数量为1，且各个计算节点的节点处理数据均被聚合之后，便可以确定当前队列中的数据即为集群聚合数据，便可以将该集群聚合数据发送至客户端。

在本实施例中，采用多线程方式，基于计算集群的各个计算节点，对列式存储的初始值和节点数据按列进行划分，得到各个线程对应的子数据，从而各个线程基于列式遍历的方式，读取数据并进行处理，进而整合各个线程处理后的数据，得到计算节点的节点处理数据；为了提高各个线程的运算效率，对于多个线程均需要计算的共享变量，采用预先计算的方式进行计算，进而相应的线程便可直接获取该共享变量，避免了重复计算，提高了数据处理效率；当服务器中存在多个计算节点时，先以服务器为单位对节点处理数据进行聚合，进而由聚合节点对各个服务器的服务器聚合数据进行聚合，采用并行的方式聚合，大大减少了聚合所需的时间，提高了聚合的效率；在聚合节点进行各个计算节点的节点处理数据的聚合时，采用当队列中存在至少两个数据时，便进行聚合的方式，大大减少了数据等待的时间，提高了聚合的效率。

图7是本发明另一实施例提供的数据聚合处理方法的流程图，本实施例是在图5所示实施例的基础上，对各个步骤采用的算法和执行主体进行进一步细化，结合图5和图7可知，该数据聚合处理方法由计算集群执行，该计算集群包括一个客户端和多个计算节点，客户端为执行客户的相关业务的执行方，用于执行本地化代码和来自计算集群的集群聚合数据，可以通过网络与计算集群进行连接。计算集群中多个计算节点则主要负责节点数据的存储和计算，各个计算节点的数据均采用列式存储的方式进行存储。本实施例提供的数据聚合处理方法的具体过程为：

首先，由客户端driver广播初始值zeroValue至每个计算节点。每个计算节点通过其聚合算子中的初始值获取模块，获取zeroValue和节点数据inputValue，其中，zeroValue和inputValue均采用堆外内存的存储方式，该聚合算子是改进的aggregate算子，其具体形式为Aggregate(zeroValue,commParaOp,seqOp,combOp)，其中zeroValue、commParaOp、seqOp、combOp为改进的aggregate算子的四个参数接口。针对每个计算节点，通过其聚合算子中的共享变量计算模块commParaOp，基于单线程模式，根据zeroValue和inputValue生成共享变量Comm_Map。通过其聚合算子中的逻辑操作模块seqOp，基于多线程模式，其中线程数n自动确定，将zeroValue和inputValue按列拆分成n份，但每个数据指向的内存地址不变，进而将拆分后的zeroValue、inputValue以及上一步commParaOp输出的Comm_Map作为各个线程的输入，执行用户自定义的预设逻辑算法，得到各个线程的线程处理数据，进而整合各个线程处理数据，得到该计算节点的节点处理数据，并将节点处理数据直接写入zeroValue中，即seqOp(zeroValue,inputValue,Comm_Map)＝>zeroValue，其中，seqOp方的代码采用先按列遍历，再按行遍历，且最好依次遍历完一列的原则执行。针对每个计算节点，以服务器为单位，对于属于同一个服务器的各个计算节点的节点处理数据，通过该服务器的预设计算节点的聚合算子中的聚合模块combOp，将该服务器的各个计算节点的节点处理数据聚合，得到服务器聚合数据；进而由计算集群的聚合节点的聚合算子的聚合模块combOp将各个服务器聚合数据聚合，得到集群聚合数据；最后，客户端driver从聚合节点拉取(pull)该集群聚合数据，并进行相应的业务操作。

图8是本发明实施例提供的数据聚合处理装置的结构示意图，如图8所示，该数据聚合处理装置包括：数据获取模块810、数据处理模块820、数据发送模块830和数据聚合模块840。

其中，数据获取模块810，用于获取初始值和计算节点的节点数据，其中，所述节点数据和所述初始值的存储方式为列式存储；数据处理模块820，用于基于列式遍历的方式，对所述初始值和所述节点数据进行处理，以生成节点处理数据；数据发送模块830，用于若所述计算节点为处理节点，则将所述节点处理数据发送至所述计算集群的聚合节点；数据聚合模块840，用于若所述计算节点为所述聚合节点，则将各个所述节点处理数据聚合，得到集群聚合数据，以将所述集群聚合数据发送至客户端。

可选地，数据处理模块820，包括：

数据读取单元，用于按照列式遍历的方式，读取所述计算节点的所述节点数据和所述初始值的每列数据；数据处理单元，用于基于预设逻辑算法，根据所述节点数据和所述初始值的每列数据，生成所述计算节点的所述节点处理数据。

可选地，数据读取单元，包括：

数据划分子单元，用于根据所述计算节点的线程的数量，将所述节点数据和所述初始值按列进行划分，以得到多个节点子数据和多个子初始值，其中，所述多个节点子数据的数量以及所述多个子初始值的数量均与所述线程的数量一致；数据读取子单元，用于针对所述计算节点的每个线程，基于列式遍历的方式，读取所述线程对应的节点子数据和子初始值的每列数据。

相应的，数据处理单元，具体用于：

针对所述计算节点的每个线程，基于预设逻辑算法，根据所述线程的所述节点子数据和所述子初始值的每列数据，生成所述线程的线程处理数据；根据各个所述线程处理数据生成所述计算节点的所述节点处理数据。

可选地，该数据聚合处理装置还包括：

共享变量生成模块，用于在根据所述计算节点的线程的数量，将所述节点数据和所述初始值按列进行划分之前，根据所述计算节点的所述节点数据和所述初始值，生成所述计算节点的各个线程共享的共享变量。

相应的，数据处理单元，具体用于：

针对所述计算节点的每个需要调用共享变量的线程，基于预设逻辑算法，根据所述共享变量、所述线程的所述节点子数据和所述子初始值的每列数据，生成所述线程的线程处理数据；针对所述计算节点的每个不需要调用共享变量的线程，基于预设逻辑算法，根据所述线程的所述节点子数据和所述子初始值的每列数据，生成所述线程的线程处理数据。

可选地，数据聚合模块840，具体用于：

获取第一计算节点的节点处理数据；当存在至少一个第二计算节点的节点处理数据时，则将所述第一计算节点的节点处理数据和所述至少一个第二计算节点的节点处理数据聚合，得到中间聚合数据，其中，所述第一计算节点为所述计算集群中的任意一个所述计算节点，所述第二计算节点为所述计算集群中的所述第一计算节点不同的所述计算节点；将所述中间聚合数据放置于所述节点处理数据对应的队列中，并删除所述第一计算节点的节点数据和至少一个第二计算节点的节点处理数据；当各个所述计算节点的节点处理数据均被聚合，且所述节点处理数据对应的队列中的数据的数量为1时，聚合完毕，所述节点处理数据对应的队列中的数据为所述集群聚合数据。

可选地，所述计算集群包括多个服务器，每一服务器包括至少一个计算节点；该数据聚合处理装置还包括：

服务器聚合模块，用于在生成所述计算节点的所述节点处理数据之后，若存在与所述计算节点属于同一服务器的其它计算节点，则经由所述服务器的预设计算节点，将所述服务器的各个计算节点的所述节点处理数据聚合，以得到服务器聚合数据，其中，所述预设计算节点为任意一个所述服务器的所述计算节点。

相应的，数据聚合模块840，具体用于：

经由所述聚合节点将所述计算集群的各个服务器的服务器聚合数据聚合，得到集群聚合数据。

可选地，该数据聚合处理装置还包括：

存储方式转换模块，用于在获取计算节点的节点数据之后，将所述节点数据的存储方式转换为堆外内存的存储方式。

本发明实施例所提供的数据聚合处理装置可执行本发明任意实施例所提供的数据聚合处理方法，具备执行方法相应的功能模块和有益效果。

图9为本发明一个实施例提供的计算节点的结构示意图，如图9所示，该计算节点包括：存储器910，处理器920以及计算机程序。

其中，计算机程序存储在存储器910中，并被配置为由处理器920执行以实现本发明图2-图7所对应的实施例中任意实施例提供的数据聚合处理方法。

其中，存储器910和处理器920通过总线930连接。

相关说明可以对应参见图2-图7的步骤所对应的相关描述和效果进行理解，此处不做过多赘述。

图10为本发明一个实施例提供的计算集群的结构示意图，如图10所示，该计算集群包括多个图9所示实施例提供的计算节点1010，图10中以5个作为示例进行说明，每个计算节点1010可以作为计算集群的一个计算节点，进行相应的数据处理，各个计算节点1010之间可以通过网络进行数据通讯，以完成对分布式存储的业务数据的聚合操作。

本发明一个实施例提供一种计算机可读存储介质，其上存储有计算机程序，计算机程序被处理器执行以实现本发明图2-图7所对应的实施例中任意实施例提供的数据聚合处理方法。

其中，计算机可读存储介质可以是ROM、随机存取存储器(RAM)、CD-ROM、磁带、软盘和光数据存储设备等。

本发明一个实施例还提供一种程序产品，该程序产品包括可执行指令，该可执行指令存储在可读存储介质中。计算节点的至少一个处理器可以从可读存储介质读取该执行指令，至少一个处理器执行该执行指令使得数据聚合处理装置实施上述任意实施例提供的数据聚合处理方法。

在本发明所提供的几个实施例中，应该理解到，所揭露的设备和方法，可以通过其它的方式实现。例如，以上所描述的设备实施例仅仅是示意性的，例如，所述模块的划分，仅仅为一种逻辑功能划分，实际实现时可以有另外的划分方式，例如多个模块可以结合或者可以集成到另一个系统，或一些特征可以忽略，或不执行。另一点，所显示或讨论的相互之间的耦合或直接耦合或通信连接可以是通过一些接口，装置或模块的间接耦合或通信连接，可以是电性，机械或其它的形式。

所述作为分离部件说明的模块可以是或者也可以不是物理上分开的，作为模块显示的部件可以是或者也可以不是物理单元，即可以位于一个地方，或者也可以分布到多个网络单元上。可以根据实际的需要选择其中的部分或者全部模块来实现本实施例方案的目的。

另外，在本发明各个实施例中的各功能模块可以集成在一个处理单元中，也可以是各个模块单独物理存在，也可以两个或两个以上模块集成在一个单元中。上述模块成的单元既可以采用硬件的形式实现，也可以采用硬件加软件功能单元的形式实现。

上述以软件功能模块的形式实现的集成的模块，可以存储在一个计算机可读取存储介质中。上述软件功能模块存储在一个存储介质中，包括若干指令用以使得一台计算机设备(可以是个人计算机，服务器，或者网络设备等)或处理器(英文：processor)执行本发明各个实施例所述方法的部分步骤。

应理解，上述处理器可以是中央处理单元(Central Processing Unit，简称CPU)，还可以是其他通用处理器、数字信号处理器(Digital Signal Processor，简称DSP)、专用集成电路(Application Specific Integrated Circuit，简称ASIC)等。通用处理器可以是微处理器或者该处理器也可以是任何常规的处理器等。结合发明所公开的方法的步骤可以直接体现为硬件处理器执行完成，或者用处理器中的硬件及软件模块组合执行完成。

存储器可能包含高速RAM存储器，也可能还包括非易失性存储NVM，例如至少一个磁盘存储器，还可以为U盘、移动硬盘、只读存储器、磁盘或光盘等。

总线可以是工业标准体系结构(Industry Standard Architecture，简称ISA)总线、外部设备互连(Peripheral Component，简称PCI)总线或扩展工业标准体系结构(Extended Industry Standard Architecture，简称EISA)总线等。总线可以分为地址总线、数据总线、控制总线等。为便于表示，本发明附图中的总线并不限定仅有一根总线或一种类型的总线。

上述存储介质可以是由任何类型的易失性或非易失性存储设备或者它们的组合实现，如静态随机存取存储器(SRAM)，电可擦除可编程只读存储器(EEPROM)，可擦除可编程只读存储器(EPROM)，可编程只读存储器(PROM)，只读存储器(ROM)，磁存储器，快闪存储器，磁盘或光盘。存储介质可以是通用或专用计算机能够存取的任何可用介质。

一种示例性的存储介质耦合至处理器，从而使处理器能够从该存储介质读取信息，且可向该存储介质写入信息。当然，存储介质也可以是处理器的组成部分。处理器和存储介质可以位于专用集成电路(Application Specific Integrated Circuits，简称ASIC)中。当然，处理器和存储介质也可以作为分立组件存在于电子设备或主控设备中。需要说明的是，在本文中，术语“包括”、“包含”或者其任何其他变体意在涵盖非排他性的包含，从而使得包括一系列要素的过程、方法、物品或者装置不仅包括那些要素，而且还包括没有明确列出的其他要素，或者是还包括为这种过程、方法、物品或者装置所固有的要素。在没有更多限制的情况下，由语句“包括一个……”限定的要素，并不排除在包括该要素的过程、方法、物品或者装置中还存在另外的相同要素。

上述本发明实施例序号仅仅为了描述，不代表实施例的优劣。

通过以上的实施方式的描述，本领域的技术人员可以清楚地了解到上述实施例方法可借助软件加必需的通用硬件平台的方式来实现，当然也可以通过硬件，但很多情况下前者是更佳的实施方式。基于这样的理解，本发明的技术方案本质上或者说对现有技术做出贡献的部分可以以软件产品的形式体现出来，该计算机软件产品存储在一个存储介质(如ROM/RAM、磁碟、光盘)中，包括若干指令用以使得一台终端设备(可以是手机，计算机，服务器，空调器，或者网络设备等)执行本发明各个实施例所述的方法。

以上仅为本发明的优选实施例，并非因此限制本发明的专利范围，凡是利用本发明说明书及附图内容所作的等效结构或等效流程变换，或直接或间接运用在其他相关的技术领域，均同理包括在本发明的专利保护范围内。