流式数据的分布式处理方法和装置

摘要

本申请公开了流式数据的分布式处理方法和装置。所述方法的一具体实施方式包括：将接收到的流式数据封装为第一弹性分布式数据集；按照时间窗口对所述第一弹性分布式数据集执行分组操作，分组操作包括：将第一弹性分布式数据集中的各个数据分别分入该数据所记载的时间戳所属时间窗口所对应的分组中，形成与各个时间窗口分别对应的、包含有穷个数据的第二弹性分布式数据集；将各个第二弹性分布式数据集封装成包含多个第二弹性分布式数据集的嵌套式数据集；使用预先定义的遍历算子，依次将所述嵌套式数据集中的各个第二弹性分布式数据集传递给定义在有穷数据集上的批处理算子以执行分布式数据处理。该实施方式实现了弹性分布式数据集中算子的复用。

权利要求

1.一种流式数据的分布式处理方法，其特征在于，所述方法包括：将接收到的流式数据封装为第一弹性分布式数据集；

按照时间窗口对所述第一弹性分布式数据集执行分组操作，所述分组操作包括：将所述第一弹性分布式数据集中的各个数据分别分入该数据所记载的时间戳所属时间窗口所对应的分组中，形成与各个时间窗口分别对应的、包含有穷个数据的第二弹性分布式数据集；

将各个第二弹性分布式数据集封装成包含多个第二弹性分布式数据集的嵌套式数据集；

使用预先定义的遍历算子，依次将所述嵌套式数据集中的各个第二弹性分布式数据集传递给定义在有穷数据集上的批处理算子以执行分布式数据处理。

2.根据权利要求1所述的方法，其特征在于，所述将各个第二弹性分布式数据集进行封装成包含多个第二弹性分布式数据集的嵌套式数据集，包括：将从各个时间窗口中选择的时间点作为键以及将各个第二弹性分布式数据集作为值，封装成以键值对形式存储各个第二弹性分布式数据集的嵌套式数据集；以及所述依次将所述嵌套式数据集中的各个第二弹性分布式数据集传递给定义在有穷数据集上的批处理算子以执行分布式数据处理，包括：依次将嵌套式数据集中作为值的各个第二弹性分布式数据集，依次传递给定义在有穷数据集上的批处理算子以执行分布式数据处理。

3.根据权利要求2所述的方法，其特征在于，所述将从各个时间窗口中选择的时间点作为键以及将各个第二弹性分布式数据集作为值，封装成以键值对存储各个第二弹性分布式数据集的嵌套式数据集，包括：将键值对的键设置分别为各个时间窗口的起始时间点。

4.根据权利要求1所述的方法，其特征在于，所述方法还包括：获取用户设置的步长时间，并将相邻时间窗口的时间间隔的值设置为所述步长时间。

5.根据权利要求4所述的方法，其特征在于，所述方法还包括：将各个时间窗口的时间长度的值设置为所述步长时间。

6.根据权利要求4所述的方法，其特征在于，所述方法还包括：获取用户设置的步长数，并将各个时间窗口的时间长度的值设置为所述步长数与所述步长时间的乘积。

7.一种流式数据的分布式处理装置，其特征在于，所述装置包括：第一封装单元，用于将接收到的流式数据封装为第一弹性分布式数据集；

分组单元，用于按照时间窗口对所述第一弹性分布式数据集执行分组操作，所述分组操作包括：将所述第一弹性分布式数据集中的各个数据分别分入该数据所记载的时间戳所属时间窗口所对应的分组中，形成与各个时间窗口分别对应的、包含有穷个数据的第二弹性分布式数据集；

第二封装单元，用于将各个第二弹性分布式数据集封装成包含多个第二弹性分布式数据集的嵌套式数据集；

执行单元，用于使用预先定义的遍历算子，依次将所述嵌套式数据集中的各个第二弹性分布式数据集传递给定义在有穷数据集上的批处理算子以执行分布式数据处理。

8.根据权利要求7所述的装置，其特征在于，所述第二封装单元进一步用于：将从各个时间窗口中选择的时间点作为键以及将各个第二弹性分布式数据集作为值，封装成以键值对形式存储各个第二弹性分布式数据集的嵌套式数据集；以及所述执行单元进一步用于：依次将嵌套式数据集中作为值的各个第二弹性分布式数据集，依次传递给定义在有穷数据集上的批处理算子以执行分布式数据处理。

9.根据权利要求8所述的装置，其特征在于，所述执行单元进一步用于：将键值对的键设置分别为各个时间窗口的起始时间点。

10.根据权利要求7所述的装置，其特征在于，所述装置还包括：窗口间隔设置单元，用于获取用户设置的步长时间，并将相邻时间窗口的时间间隔的值设置为所述步长时间。

11.根据权利要求10所述的装置，其特征在于，所述装置还包括：窗口时长设置单元，用于将各个时间窗口的时间长度的值设置为所述步长时间。

12.根据权利要求10所述的装置，其特征在于，所述装置还包括：窗口时长设置单元，用于获取用户设置的步长数，并将各个时间窗口的时间长度的值设置为所述步长数与所述步长时间的乘积。

说明书

流式数据的分布式处理方法和装置

技术领域

[0001] 本申请涉及计算机技术领域，具体涉及大数据处理技术领域，尤其涉及流式数据的分布式处理方法和装置。

背景技术

[0002] 随着大数据量计算技术的发展，基于数据处理的应用受到广泛关注。数据源的结构显示出多样化的趋势，数据源产生的数据中不仅有传统的非实时的、静态结构化数据，还有很多实时的、动态产生的非结构化数据流，从这类连续到达的非结构化数据序列中获取携带的重要信息，需要依靠分布式流式数据实时计算技术。

[0003] 目前，常用的流式数据处理的框架包括Storm框架和Spark框架，分布式实时计算的表示层的高级封装则有对应的Storm原生接口以及弹性分布式数据集(Resilient Distributed Datasets，RDD)。其中Storm原生接口的封装程度较低，接口抽象程度较低，接口使用较为不便，实现时间窗口十分复杂，代码可重用性较差；使用弹性分布式数据集接口则面临强制指定以数据流入时间驱动时间窗口、不支持嵌套、无法复用代码以及批处理和流式计算不能保证接口兼容等问题。

发明内容

[0004] 本申请的目的在于提出一种改进的流式数据的分布式处理方法和装置，来解决以上背景技术部分提到的技术问题。

[0005] 第一方面，本申请提供了一种流式数据的分布式处理方法，所述方法包括：将接收到的流式数据封装为第一弹性分布式数据集；按照时间窗口对所述第一弹性分布式数据集执行分组操作，所述分组操作包括：将所述第一弹性分布式数据集中的各个数据分别分入该数据所记载的时间戳所属时间窗口所对应的分组中，形成与各个时间窗口分别对应的、包含有穷个数据的第二弹性分布式数据集；将各个第二弹性分布式数据集封装成包含多个第二弹性分布式数据集的嵌套式数据集；使用预先定义的遍历算子，依次将所述嵌套式数据集中的各个第二弹性分布式数据集传递给定义在有穷数据集上的批处理算子以执行分布式数据处理。

[0006] 在一些实施例中，所述将各个第二弹性分布式数据集进行封装成包含多个第二弹性分布式数据集的嵌套式数据集，包括：将从各个时间窗口中选择的时间点作为键以及将各个第二弹性分布式数据集作为值，封装成以键值对形式存储各个第二弹性分布式数据集的嵌套式数据集；以及所述依次将所述嵌套式数据集中的各个第二弹性分布式数据集传递给定义在有穷数据集上的批处理算子以执行分布式数据处理，包括：依次将嵌套式数据集中作为值的各个第二弹性分布式数据集，依次传递给定义在有穷数据集上的批处理算子以执行分布式数据处理。

[0007] 在一些实施例中，所述将从各个时间窗口中选择的时间点作为键以及将各个第二弹性分布式数据集作为值，封装成以键值对存储各个第二弹性分布式数据集的嵌套式数据集，包括：将键值对的键设置分别为各个时间窗口的起始时间点。

[0008] 在一些实施例中，所述方法还包括：获取用户设置的步长时间，并将相邻时间窗口的时间间隔的值设置为所述步长时间。

[0009] 在一些实施例中，所述方法还包括：所述方法还包括：将各个时间窗口的时间长度的值设置为所述步长时间。

[0010] 在一些实施例中，所述方法还包括：获取用户设置的步长数，并将各个时间窗口的时间长度的值设置为所述步长数与所述步长时间的乘积。

[0011] 第二方面，本申请提供了一种流式数据的分布式处理装置，所述装置包括：第一封装单元，用于将接收到的流式数据封装为第一弹性分布式数据集；分组单元，用于按照时间窗口对所述第一弹性分布式数据集执行分组操作，所述分组操作包括：将所述第一弹性分布式数据集中的各个数据分别分入该数据所记载的时间戳所属时间窗口所对应的分组中，形成与各个时间窗口分别对应的、包含有穷个数据的第二弹性分布式数据集；第二封装单元，用于将各个第二弹性分布式数据集封装成包含多个第二弹性分布式数据集的嵌套式数据集；执行单元，用于使用预先定义的遍历算子，依次将所述嵌套式数据集中的各个第二弹性分布式数据集传递给定义在有穷数据集上的批处理算子以执行分布式数据处理。

[0012] 在一些实施例中，所述第二封装单元进一步用于：将从各个时间窗口中选择的时间点作为键以及将各个第二弹性分布式数据集作为值，封装成以键值对形式存储各个第二弹性分布式数据集的嵌套式数据集；以及所述执行单元进一步用于：依次将嵌套式数据集中作为值的各个第二弹性分布式数据集，依次传递给定义在有穷数据集上的批处理算子以执行分布式数据处理。

[0013] 在一些实施例中，所述执行单元进一步用于：将键值对的键设置分别为各个时间窗口的起始时间点。

[0014] 在一些实施例中，所述装置还包括：窗口间隔设置单元，用于获取用户设置的步长时间，并将相邻时间窗口的时间间隔的值设置为所述步长时间。

[0015] 在一些实施例中，所述装置还包括：窗口时长设置单元，用于将各个时间窗口的时间长度的值设置为所述步长时间。

[0016] 在一些实施例中，所述装置还包括：窗口时长设置单元，用于获取用户设置的步长数，并将各个时间窗口的时间长度的值设置为所述步长数与所述步长时间的乘积。

[0017] 本申请提供的流式数据的分布式处理方法和装置，通过对流式数据封装成的弹性分布式数据集按时间窗口进行分组，并根据各个时间窗口对应的第二弹性分布式数据集生成嵌套式数据集，再利用预先定义的算子对嵌套式数据集中的各个第二弹性分布式数据集分别处理，由于嵌套式数据集中的单个第二弹性分布式数据集均为包含有穷个数据的有穷数据集，因此可以复用已有的算子，提高了算子的复用性，减少了设计额外算子的开发成本。另外，本方案可以在批处理和流式不同引擎可以使用相同的代码实现相同的计算语意，可以减少切换不同引擎时修改代码所需的工作量。

附图说明

[0018] 通过阅读参照以下附图所作的对非限制性实施例所作的详细描述，本申请的其它特征、目的和优点将会变得更明显：

[0019] 图1是本申请可以应用于其中的示例性系统架构图；

[0020] 图2是根据本申请的流式数据的分布式处理方法的一个实施例的流程图；

[0021] 图3是根据本申请的流式数据的分布式处理的又一个实施例的流程图；

[0022] 图4是根据本申请的流式数据的分布式处理装置的一个实施例的结构示意图；

[0023] 图5是适于用来实现本申请实施例的终端设备或服务器的计算机系统的结构示意图。

具体实施方式

[0024] 下面结合附图和实施例对本申请作进一步的详细说明。可以理解的是，此处所描述的具体实施例仅仅用于解释相关发明，而非对该发明的限定。另外还需要说明的是，为了便于描述，附图中仅示出了与有关发明相关的部分。

[0025] 需要说明的是，在不冲突的情况下，本申请中的实施例及实施例中的特征可以相互组合。下面将参考附图并结合实施例来详细说明本申请。

[0026] 图1示出了可以应用本申请的用于流式数据的批处理方法或装置的实施例的示例性系统架构100。

[0027] 如图1所示，系统架构100可以包括驱动节点101和执行节点102、103、104。驱动节点101和执行节点102、103、104之间可以通过网络进行通信，网络可以包括各种连接类型，例如有线、无线通信链路或者光纤电缆等等。

[0028] 驱动节点101可以将对数据的处理任务分发给执行节点102、103、104执行，各执行节点负责具体执行分发到的处理任务。

[0029] 需要说明的是，本申请实施例所提供的流式数据的分布式处理方法一般由驱动节点101执行，一些步骤也可以由执行节点102、103、104执行；相应地，流式数据的分布式处理装置一般设置于驱动节点101中，一些单元也可以设置于执行节点102、103、104中。

[0030] 应该理解，图1中的驱动节点和执行节点的数目仅仅是示意性的。根据实现需要，可以具有任意数目的驱动节点和执行节点。

[0031] 继续参考图2，示出了根据本申请的数据流的分布式处理方法的一个实施例的流程200。所述的数据流的分布式处理方法，包括以下步骤：

[0032] 步骤201，将接收到的流式数据封装为第一弹性分布式数据集。

[0033] 在本实施例中，数据流的分布式处理方法运行于其上的电子设备(例如图1所示的驱动节点101)可以通过有线连接方式或者无线连接方式从不断生成数据的流式数据源接收流式数据。需要指出的是，上述无线连接方式可以包括但不限于3G/4G连接、WiFi连接、蓝牙连接、WiMAX连接、Zigbee连接、UWB(ultra wideband)连接、以及其他现在已知或将来开发的无线连接方式。所接收的流式数据中的各个数据可以包括时间戳，该时间戳用于标识该数据的时间。可选的，该时间可以是数据流入时间，也可以是其他与数据相关的时间(例如数据生成时间)。对于接收到的流式数据，电子设备可以对其进行封装处理，将流式数据封装成弹性分布式数据集中的形式，形成第一弹性分布式数据集，以便后续执行进一步的处理。由于流式数据中各个数据包含相应的时间戳，封装后的第一弹性分布式数据集中各个数据又具有时间戳。通常，由于流式数据可以是源源不断产生的，封装而成的第一弹性分布式数据集中可能包括无数个数据。

[0034] 步骤202，按照时间窗口对第一弹性分布式数据集中的数据执行分组操作。

[0035] 在本实施例中，基于步骤201中所生成的第一弹性分布式数据集，电子设备可以首先按照时间窗口对第一弹性分布式数据集进行分组操作。在分组操作时，可以按照时间戳与时间窗口的对应关系，将数据分入其时间戳所属时间窗口所对应的分组中，从而使得分组操作后形成各个时间窗口对应的第二弹性分布式数据集。时间窗口的时间长度可以由系统自动设置，也可以由用户进行设置。通常，第一弹性分布式数据集可能包含无数个数据，且由于时间窗口的无限性，因此，分组操作可以形成无数个第二弹性分布式数据集。但是，由于单个时间窗口的时间长度是有限的，因此各个时间窗口对应的第二弹性分布式数据集包含有穷个数据，即该第二弹性分布式数据集可以是有穷数据集。

[0036] 步骤203，将各个第二弹性分布式数据集封装成包含多个第二弹性分布式数据集的嵌套式数据集。

[0037] 在本实施例中，基于步骤202形成的、分别与各个时间窗口对应的第二弹性分布式数据集，电子设备可以将各个第二弹性分布式数据集聚合起来，并封装成构建成包含多个第二弹性分布式数据集的嵌套式数据集。所谓嵌套式数据集，即该数据集中的各个元素还是数据集。

[0038] 步骤204，使用预先定义的遍历算子，依次将嵌套式数据集中的各个第二弹性分布式数据集传递给定义在有穷数据集上的批处理算子以执行分布式数据处理。

[0039] 在本实施例中，基于步骤203所生成的嵌套式数据集，电子设备可以对其进行进一步处理。如上所述，由于流式数据可以是源源不断的数据，因此时间窗可以是无限的，步骤203所形成的嵌套式数据集中也可以包括无穷个第二弹性分布式数据集，但各个第二弹性分布式数据集均是包括有穷个元素的数据集。因此，可以使用预先定义在嵌套式数据集上的遍历算子，依次将上述嵌套式数据集的第二弹性分布式数据集传递至定义在有穷数据集上的批处理算子，从而使得批处理算子分别对各个嵌套式数据集进行分布式处理。其中，上述遍历算子在Java语言中可以是apply或相关的方法。上述批处理算子是可以对有穷数据集进行处理的各种函数，例如聚类算子kmeans。

[0040] 在本实施例的一些可选实现方式中，步骤203可以进一步包括：将从各个时间窗口中选择的时间点作为键以及将各个第二弹性分布式数据集作为值，封装成以键值对形式存储各个第二弹性分布式数据集的嵌套式数据集；步骤204中依次将嵌套式数据集中的各个第二弹性分布式数据集传递给定义在有穷数据集上的批处理算子以执行分布式数据处理，可以具体包括：依次将嵌套式数据集中作为值的各个第二弹性分布式数据集，依次传递给定义在有穷数据集上的批处理算子以执行分布式数据处理。在本实施例中，封装成的嵌套数据集以键值对形式存储第二弹性分布式数据集，其中键的部分设置为各个时间窗口的范围内的一个时间点，值的部分用于存储各个第二弹性分布式数据集。作为键的各个时间窗口的时间点可以是按照统一的标准设定，例如可以是各个时间窗的中点时间点或终止时间点。在利用遍历算子处理嵌套数据集时，只需要通过用于对嵌套式数据集的值进行遍历的遍历算子，并结合用于对普通的弹性分布式数据集(包含有穷个元素)的批处理算子，即可分别对嵌套式数据集中的各个第二弹性分布式数据集进行处理。在该实现方式中，用于对嵌套式数据集的值进行遍历的遍历算子通常可以写作apply_values。

[0041] 实践中，步骤202、步骤203以及步骤204的操作可以通过Java语言代码rdd.window_into(Window(1 hour)).apply_values(kmeans)触发。其中，rdd可以是步骤201封装而成的第一弹性分布式数据集。通过window_into(Window(1 hour))可以使第一弹性分布式数据集按一个小时划分时间窗口，以对第一弹性分布式数据集进行数据分组形成各个第二弹性分布式数据集，并形成嵌套式数据集。apply_values是用于遍历嵌套式数据集中各个值的遍历算子，该遍历算子将第二弹性分布式数据集中的各个值(即第二弹性分布式数据集)传递给聚类算子kmeans算子进行处理，kmeans算子即可分别对第二弹性分布式数据集进行聚类处理。需要说明的是，以上代码仅为示例，实践中还可以通过其他代码执行以上过程。

[0042] 在本实施例的一些可选实现方式中，键值对数据的键名部分可以统一设置为时间窗口的起始时间点。

[0043] 本申请的上述实施例提供的方法通过对流式数据封装成的弹性分布式数据集按时间窗口进行分组，并根据各个时间窗口对应的第二弹性分布式数据集生成嵌套式数据集，再利用预先定义的算子对嵌套式数据集中的各个第二弹性分布式数据集分别处理，由于嵌套式数据集中的单个第二弹性分布式数据集均为包含有限个数据的有穷数据集，因此可以复用已有的算子，提高了算子的复用性，减少了设计额外算子的开发成本。另外，当切换到批处理引擎时，也可以以相同的方案进行时间窗口对每次的批数据进行分组操作并形成包括多个有穷数据集的嵌套式数据集，并使用遍历算法分别将有穷数据集传递给批处理算子进行操作，这使得本方案可以在批处理和流式不同引擎可以使用相同的代码实现相同的计算语意，可以减少不同处理引擎切换时修改代码所需要的工作量。

[0044] 进一步参考图3，其示出了流式数据的分布式处理方法的又一个实施例的流程300。该流式数据的分布式处理方法的流程300，包括以下步骤：

[0045] 步骤301，将接收到的流式数据封装为第一弹性分布式数据集。

[0046] 在本实施例中，步骤301的具体处理可以参考图2对应实施例的步骤201。

[0047] 步骤302，获取用户设置的步长时间，并将相邻时间窗口的时间间隔的值设置为该步长时间。

[0048] 在本实施例中，电子设备可以获取用户设置的步长时间。实践中，用户设置的步长时间可以作为参数传入，以使电子设备获取到该步长时间并根据该步长时间进行设置处理，以将相邻时间窗口的时间间隔设置为该步长时间。

[0049] 需要说明的是，相邻时间窗口在时间上可以是相连的，也可以是重叠的。相邻时间窗口在时间上相连时，当前时间窗口的开始时间点和上一个时间窗口的结束时间点可以是连续的。例如，当最小时间单位为秒时，当前时间窗口的开始时间点为5秒，前一个时间窗口的结束时间点可以是4秒。当相邻时间窗口在时间上重叠时，当前时间窗口的开始时间可以是前一个时间窗口内部的一个时间点。例如当前时间窗口为2秒到6秒，前一个时间窗口的时间可以是0秒到4秒，当前时间窗口的开始时间2秒处于0秒到4秒之间。

[0050] 相邻时间窗口的时间间隔也可以称为步长值，即该步长值可以由用户进行设置。时间窗口的长度通常可以是步长值的整数倍，该倍数可以称为步长数。

[0051] 步骤303，按照时间窗口对第一弹性分布式数据集中的数据执行分组操作。

[0052] 在本实施例中，步骤303的具体处理可以参考图2对应实施例中的步骤202，这里不再赘述。

[0053] 步骤304，将各个第二弹性分布式数据集封装成包含多个第二弹性分布式数据集的嵌套式数据集。

[0054] 在本实施例中，步骤304的具体处理可以参考图2对应实施例中的步骤203，这里不再赘述。

[0055] 步骤305，使用预先定义的遍历算子，依次将嵌套式数据集中的各个第二弹性分布式数据集传递给定义在有穷数据集上的批处理算子以执行分布式数据处理。

[0056] 在本实施例中，步骤305的具体处理可以参考图2对应实施例中的步骤204，这里不再赘述。

[0057] 在本实施例的一些可选实现方式中，可以将各个时间窗口的时间长度的值设置为该步长时间。在该实现方式中，当用户没有对时间窗口的时间长度特别设置时，电子设备直接将上述步长时间(即相邻时间窗口的时间间隔的值)也确定为时间窗口的时间长度。此时，相邻时间窗口在时间上是相连。

[0058] 在本实施例的一些可选实现方式中，上述流式数据的分布式处理方法还包括以下步骤：获取用户设置的步长数；将各个时间窗口的时间长度的值设置为步长数与步长时间的乘积。在该实现方式中，用户可以自行设置步长数。电子设备可以获取该步长数，并将该步长数与步长时间相乘，得到的时间可以作为各个时间窗口的时间长度。

[0059] 在实践中，用户可以将用于执行分组操作与封装嵌套式数据集所使用的java代码设置为window_into(Window(5s，1))，其中5s是用户设置的步长时间5s，电子设备可以将5s设置为相邻时间窗口的间隔。1是用户设置的步长数，电子设备可以将5s与1的乘积(仍是5s)设为时间窗口的时间长度。各个时间窗口依次是0-4s、5-9s、10-14s……。假设数据流为(1s,1),(9s,2),(11s,3),(8s,4),(2s,5),(20s,6)…，在通过上述设置的方法对数据流封装后的第一弹性分布式数据集进行处理时，所形成的嵌套式数据集可以表示为：

[0060]

[0061] 之后，可以通过遍历算子结合批处理算子对嵌套式数据集中的各个第二弹性分布式数据集(例如[(1,1s),(5,2s)]、[(2,9s),(4,8s)])分别执行分布式数据处理。

[0062] 从图3中可以看出，与图2对应的实施例相比，本实施例中流式数据的分布式处理方法的流程300突出了按照用户设置的步长时间设置时间窗口的间隔，提高了操作的灵活性，从而能适应更广泛的应用场景。

[0063] 进一步参考图4，作为对上述各图所示方法的实现，本申请提供了一种流式数据的分布式处理装置的一个实施例，该装置实施例与图2所示的方法实施例相对应，该装置具体可以应用于各种电子设备中。

[0064] 如图4所示，本实施例所述的流式数据的分布式处理装置400包括：第一封装单元401、分组单元402、第二封装单元403和执行单元404。其中，第一封装单元401用于将接收到的流式数据封装为第一弹性分布式数据集；分组单元402用于按照时间窗口对第一弹性分布式数据集执行分组操作，该分组操作包括：将第一弹性分布式数据集中的各个数据分别分入该数据所记载的时间戳所属时间窗口所对应的分组中，形成与各个时间窗口分别对应的、包含有穷个数据的第二弹性分布式数据集；第二封装单元403用于将各个第二弹性分布式数据集封装成包含多个第二弹性分布式数据集的嵌套式数据集；而执行单元404用于使用预先定义的遍历算子，依次将嵌套式数据集中的各个第二弹性分布式数据集传递给定义在有穷数据集上的批处理算子以执行分布式数据处理。

[0065] 在本实施例中，流式数据的分布式处理装置400的第一封装单元401、分组单元402、第二封装单元403和执行单元404的具体处理可以分别参考图2对应实施例的步骤201、步骤202、步骤203以及步骤204，这里不再赘述。

[0066] 在本实施例的一些可选实现方式中，第二封装单元403进一步用于：将从各个时间窗口中选择的时间点作为键以及将各个第二弹性分布式数据集作为值，封装成以键值对形式存储各个第二弹性分布式数据集的嵌套式数据集；以及执行单元404进一步用于：依次将嵌套式数据集中作为值的各个第二弹性分布式数据集，依次传递给定义在有穷数据集上的批处理算子以执行分布式数据处理。该实现方式的具体处理可以参考图2对应实施例中相应的实现方式，这里不再赘述。

[0067] 在本实施例的一些可选实现方式中，上述执行单元404进一步用于：将键值对的键设置分别为各个时间窗口的起始时间点。该实现方式的具体处理可以参考图2对应实施例中相应的实现方式，这里不再赘述。

[0068] 在本实施例的一些可选实现方式中，上述装置400还包括窗口间隔设置单元(未示出)，用于获取用户设置的步长时间，并将相邻时间窗口的时间间隔的值设置为该步长时间。该实现方式的具体处理可以参考图3对应的实施例，这里不再赘述。

[0069] 在本实施例的一些可选实现方式中，上述装置400还包括窗口时长设置单元(未示出)，用于将各个时间窗口的时间长度的值设置为上述步长时间。该实现方式的具体处理可以参考图3对应实施例中相应的实现方式，这里不再赘述。

[0070] 在本实施例的一些可选实现方式中，上述装置400还包括窗口时长设置单元(未示出)，用于获取用户设置的步长数，并将各个时间窗口的时间长度的值设置为步长数与步长时间的乘积。该实现方式的具体处理可以参考图3对应实施例中相应的实现方式，这里不再赘述。

[0071] 下面参考图5，其示出了适于用来实现本申请实施例的驱动节点或执行节点的计算机系统500的结构示意图。

[0072] 如图5所示，计算机系统500包括中央处理单元(CPU)501，其可以根据存储在只读存储器(ROM)502中的程序或者从存储部分508加载到随机访问存储器(RAM)503中的程序而执行各种适当的动作和处理。在RAM 503中，还存储有系统500操作所需的各种程序和数据。CPU 501、ROM 502以及RAM 503通过总线504彼此相连。输入/输出(I/O)接口505也连接至总线504。

[0073] 以下部件连接至I/O接口505：输入部分506；输出部分507；包括硬盘等的存储部分508；以及包括诸如LAN卡、调制解调器等的网络接口卡的通信部分509。通信部分509经由诸如因特网的网络执行通信处理。驱动器510也根据需要连接至I/O接口505。可拆卸介质511，诸如磁盘、光盘、磁光盘、半导体存储器等等，根据需要安装在驱动器510上，以便于从其上读出的计算机程序根据需要被安装入存储部分508。

[0074] 特别地，根据本公开的实施例，上文参考流程图描述的过程可以被实现为计算机软件程序。例如，本公开的实施例包括一种计算机程序产品，其包括有形地包含在机器可读介质上的计算机程序，所述计算机程序包含用于执行流程图所示的方法的程序代码。在这样的实施例中，该计算机程序可以通过通信部分509从网络上被下载和安装，和/或从可拆卸介质511被安装。

[0075] 附图中的流程图和框图，图示了按照本申请各种实施例的系统、方法和计算机程序产品的可能实现的体系架构、功能和操作。在这点上，流程图或框图中的每个方框可以代表一个模块、程序段、或代码的一部分，所述模块、程序段、或代码的一部分包含一个或多个用于实现规定的逻辑功能的可执行指令。也应当注意，在有些作为替换的实现中，方框中所标注的功能也可以以不同于附图中所标注的顺序发生。例如，两个接连地表示的方框实际上可以基本并行地执行，它们有时也可以按相反的顺序执行，这依所涉及的功能而定。也要注意的是，框图和/或流程图中的每个方框、以及框图和/或流程图中的方框的组合，可以用执行规定的功能或操作的专用的基于硬件的系统来实现，或者可以用专用硬件与计算机指令的组合来实现。

[0076] 描述于本申请实施例中所涉及到的单元可以通过软件的方式实现，也可以通过硬件的方式来实现。所描述的单元也可以设置在处理器中，例如，可以描述为：一种处理器包括第一封装单元、分组单元、第二封装单元和生成单元。其中，这些单元的名称在某种情况下并不构成对该单元本身的限定，例如，第一封装单元还可以被描述为“将接收到的流式数据封装为第一弹性分布式数据集的单元”。

[0077] 作为另一方面，本申请还提供了一种非易失性计算机存储介质，该非易失性计算机存储介质可以是上述实施例中所述装置中所包含的非易失性计算机存储介质；也可以是单独存在，未装配入终端中的非易失性计算机存储介质。上述非易失性计算机存储介质存储有一个或者多个程序，当所述一个或者多个程序被一个设备执行时，使得所述设备：将接收到的流式数据封装为第一弹性分布式数据集；按照时间窗口对所述第一弹性分布式数据集执行分组操作，所述分组操作包括：将所述第一弹性分布式数据集中的各个数据分别分入该数据所记载的时间戳所属时间窗口所对应的分组中，形成与各个时间窗口分别对应的、包含有穷个数据的第二弹性分布式数据集；将各个第二弹性分布式数据集封装成包含多个第二弹性分布式数据集的嵌套式数据集；使用预先定义的遍历算子，依次将所述嵌套式数据集中的各个第二弹性分布式数据集传递给定义在有穷数据集上的批处理算子以执行分布式数据处理。

[0078] 以上描述仅为本申请的较佳实施例以及对所运用技术原理的说明。本领域技术人员应当理解，本申请中所涉及的发明范围，并不限于上述技术特征的特定组合而成的技术方案，同时也应涵盖在不脱离所述发明构思的情况下，由上述技术特征或其等同特征进行任意组合而形成的其它技术方案。例如上述特征与本申请中公开的(但不限于)具有类似功能的技术特征进行互相替换而形成的技术方案。