CN102375877A

CN102375877A - 使用属性特征化预测性地移除运行时数据的方法和系统

Info

Publication number: CN102375877A
Application number: CN2011102387994A
Authority: CN
Inventors: M.J.布兰森; J.M.桑托索索
Original assignee: International Business Machines Corp
Current assignee: International Business Machines Corp
Priority date: 2010-08-20
Filing date: 2011-08-19
Publication date: 2012-03-14
Also published as: US20120047505A1; US8639809B2; JP2012043409A; JP5679917B2

Abstract

提供了用于处理数据流的计算机实现的方法和系统，其中，说明了以在基于流的应用中减小处理数据引起的任何延迟的影响的方式而从基于流的应用中选择性地移除运行时数据的技术。除了从所述基于流的应用的主处理路径中移除数据之外，可以以替换的方式处理所述数据，要么使用替换的处理资源，要么通过延迟所述数据的处理。

Description

使用属性特征化预测性地移除运行时数据的方法和系统

技术领域

本发明涉及数据处理，特别地，涉及使用属性特征化预测性地移除运行时数据的方法和系统。

背景技术

当计算机数据库变得极其复杂时，施加于数据库系统上的计算命令也迅速增加。数据库系统典型地被配置为将存储数据的处理与访问、操作或使用存储在数据库中的数据相分离。更具体地，数据库使用数据先被存储、接着被索引、然后被查询的模型。然而，这个模型不能满足一些实时应用的性能要求。例如，数据库系统能够接收并存储传入的数据的速率会限制可以被处理或者被评估的数据的数量。这继而限制了数据库应用实时处理大量数据的能力。

发明内容

本发明的实施例提供了一种用于处理数据流的计算机实现的方法、系统和计算机程序产品。所述方法、系统和计算机程序产品包括接收要处理的数据流的第一元组，其中所述第一元组包括多个属性。所述方法、系统和计算机程序产品还包括至少部分基于所述第一元组中的多个属性中的一个或多个属性，计算用于处理所述数据流的所接收到的第一元组的估计处理时间。另外，所述方法、系统和计算机程序产品包括：当确定所计算的估计处理时间不超过预定的阈值时间时，使用第一执行路径以第一方式处理所述数据流的第一元组。所述方法、系统和计算机程序产品还包括：当确定所计算的估计处理时间超过所述预定的阈值时间时，以第二方式处理所述数据流的所接收到的第一元组，相对于以第一方式处理所述第一元组，所述第二方式降低所述第一执行路径中的处理延迟。

本发明的替代实施例提供了一种用于处理数据流的方法、系统和计算机程序产品。所述方法、系统和计算机程序产品包括接收要处理的数据流的第一元组，其中所述第一元组包括多个属性。所述方法、系统和计算机程序产品还包括基于(i)所述第一元组中的多个属性中的一个或多个属性和(ii)与所接收到的元组相关联的元数据两者中的至少一个，计算用于处理所述数据流的所接收到的第一元组的估计处理时间。另外，所述方法、系统和计算机程序产品包括：当确定所计算的估计处理时间不超过预定阈值时间时，使用第一执行路径以第一方式处理所述数据流的第一元组。所述方法、系统和计算机程序产品还包括：当确定所计算的估计处理时间超过预定阈值时间时，以第二方式处理所述数据流的所接收到的第一元组，相对于以第一方式处理所述第一元组，所述第二方式降低所述第一执行路径中的处理延迟。

附图说明

由此，获得并可以详细理解上述方面的方式、上面简要概括的本发明的实施例的更具体的说明，可以参考附图。

然而，要注意，所述附图仅仅说明了本发明的典型实施例，因此不被认为是对它的范围的限制，本发明可以认可其他同样有效的实施例。

图1A-1B说明了根据本发明的一个实施例的、被配置用于执行流应用的计算基础结构100。

图2是根据本发明的一个实施例的、图1A-1B的计算节点130的更详细的视图。

图3是根据本发明的一个实施例的、图1的服务器计算系统105的更详细的视图。

图4说明了根据本发明的一个实施例的、流应用中的计算节点的示例。

图5是说明了根据本发明的一个实施例的、预测性地移除运行时数据的方法的流程图。

图6是说明了根据本发明的一个实施例的、概要分析(profiling)元组的方法的流程图。

具体实施方式

基于流的计算和基于流的数据库计算随着数据库系统技术的发展而出现。可以获得允许用户在流数据到达数据库文件之前创建处理和查询流数据的应用的产品。利用该出现的技术，当传入(inbound)的数据记录“在飞行中(in flight)”时，用户能够指定处理逻辑以应用到传入的数据记录，可以在非常短的时间内得到结果，通常是毫秒级的。使用这种类型的处理构建应用开辟了允许多样性的创新应用、系统和进程(process)得以开发的新编程范例，并且对应用程序员和数据库开发人员提出了新的挑战。

在流应用中，操作码(operator)相互连接使得数据从一个处理元素流向下一个(如，在TCP/IP套接字上)。通过创建许多小的可执行代码片(操作码)而在节点之间分布应用、以及通过在多个节点上复制处理元素并在它们之间进行负载平衡，来达到可扩展性(scalability)。流应用中的处理元素(以及操作码)可以融合在一起形成更大的处理元素。这样做允许处理元素共享共同的处理空间，从而在操作码之间产生比使用进程间通信技术(例如，使用TCP/IP套接字)可获得的通信快得多的通信。此外，可以从表示经过流应用的数据流的操作码图中动态地移除或插入处理元素，并且可以在运行时期间将其从流应用中解融合或融合。

流应用的一个优势是它们允许用户细粒度地控制经过所述应用的数据的处理流。换言之，用户可以为对进入数据执行不同操作的各个处理元素指派具体的操作码，并可以通过修改所述操作码以及它们被执行的顺序来动态地改变流应用。此外，流应用能够在限制处理中的任意“瓶颈”的同时处理大量的数据。然而，在很多情况下，瓶颈将仍然出现，其可能引起流应用的部分减速。此外，这样的瓶颈导致的问题可能会随着额外数据的接收而继续加重。如此，对于必须尽可能接近实时地处理进入数据的流应用，任何潜在的瓶颈都应该被避免，因为它们可能会严重影响应用的性能以及应用的实时处理进入数据的能力。

本发明的实施例提供了用于识别可能不能在阈值时间量内处理完的进入流数据并且将所述进入数据从主执行路径中移除的技术。特别地，本发明的实施例提供了用于计算针对所接收到的数据流的元组的估计处理时间、并且基于所计算的估计处理时间确定是否处理所接收到的元组的技术。例如，假定给定的处理元素处理元组的平均处理时间是5毫秒。如果本发明的实施例接收到特定的元组并计算出给定的处理元素大约需要5分钟处理所述特定元组，则本发明的实施例可以接着阻止给定的处理元素处理所述元组。如此，本发明的实施例可以防止流应用中的瓶颈。另外，在某些实施例中，可以接着使用所述主要执行路径之外的其他资源处理所述特定元组。

下面，参考本发明的实施例。然而，应该理解，本发明不局限于具体说明的实施例。相反，下面的特征和元素的任意组合(不管是否与不同的实施例相关)在实现和实践本发明时都被考虑在内。此外，虽然本发明的实施例可以达到优于其他可能的解决方案和/或现有技术的优点，但给定实施例是否达到特定优势并不是对本发明的限制。因此，以下方面、特征、实施例和优势仅仅是说明性的，除了在权利要求书中显式声明之外，并不认为是所附权利要求的元素或限制。同样地，对“本发明”的引用不应该被解释为对这里公开的任意发明主题的概括，并且除了在权利要求书中显式声明之外，也不应该被认为是所述权利要求的元素或限制。

本领域的技术人员将理解，本发明的方面可以被实施为系统、方法或计算机程序产品。因此，本发明的方面可以采用完全硬件实施例、完全软件实施例(包括固件、常驻软件、微代码，等等)或将软件和硬件方面组合的实施例的形式，这里，它们全部被一般地称为“电路”、“模块”或“系统”。此外，本发明的方面可以采用被实施在一个或多个计算机可读介质中的计算机程序产品的形式，所述计算机可读介质具有在其上包含的计算机可读程序。

可以利用一个或多个计算机可读介质的任意组合。计算机可读介质可以是计算机可读信号介质或计算机可读存储介质。计算机可读存储介质可以是，例如但不限于，电子的、磁性的、光学的、电磁的、红外的或半导体的系统、装置或器件，或前述的任意合适的组合。所述计算机可读存储介质更具体的示例(非穷举列表)可以包括以下：具有一个或多个导线的电连接；便携式计算机小磁盘；硬盘；随机存取存储器(RAM)；只读存储器(ROM)；可擦除可编程只读存储器(EPROM或闪存)；光纤；便携式光盘只读存储器(CD-ROM)；光学存储器件；磁体存储器件；或前述的任意合适的组合。在这个文件的语境中，计算机可读存储介质可以是任意可以包含或存储由指令执行系统、装置或设备使用的或与指令执行系统、装置或设备关联的程序的有形介质。

计算机可读信号介质可以包括具有在其中具体化计算机可读程序代码的传播数据信号，例如，在基带中或作为载波的一部分。这样的传播数据信号可以采用多种不同的形式，包括但不限于，电磁的、光学的或其任意合适的组合。计算机可读信号介质可以是任意计算机可读介质，其不是计算机可读存储介质且能够传递、传播或传输由指令执行系统、装置或设备使用的或与指令执行系统、装置或设备关联的程序。

包含在计算机可读介质上的程序代码可以使用任意适当的介质传送，包括但不限于无线、有线、光纤光缆、RF等等，或前述的任意合适的组合。

用于执行本发明的方面的操作的计算机程序代码可以用一个或多个编程语言的任意组合编写，包括诸如Java、Smalltalk、C++等面向对象编程语言和诸如C编程语言或类似的编程语言的传统过程化编程语言。所述程序代码可以完全在用户的计算机上、或部分在用户的计算机上作为单机软件包执行，或者部分在用户的计算机上而部分在远程计算机上或完全在远程计算机或服务器上执行。在后一情形中，所述远程计算机可以通过任意类型的网络连接到用户的计算机，包括局域网(LAN)或广域网(WAN)，或者可以连接到外部计算机(例如，使用因特网服务提供商通过因特网)。

下面将参考根据本发明的实施例的方法、装置(系统)和计算机程序产品的流程图和/或框图说明当前发明的方面。将理解，所述流程图和/或框图的各个块，以及流程图和/或框图中的块的组合，可以通过计算机程序指令实现。这些计算机程序指令可以被提供给通用计算机、专用计算机或其他可编程数据处理装置的处理器以产生机器，从而经由计算机或其他可编程数据处理装置的处理器执行的指令创建用于实现所述流程图和/或框图中指定的功能/行为的部件。

这些计算机程序指令还可以被存储在能够指引计算机、其他可编程数据处理装置或其他设备以特定的方式运行的计算机可读介质中，从而存储在计算机可读介质中的指令产生包括实现所述流程图和/或框图块中指定的功能/行为的指令的产品。

所述计算机程序指令还可以被加载到计算机、其他可编程数据处理装置或其他设备，以引起在所述计算机、其他可编程数据处理装置或其他设备上执行一系列操作步骤，以便产生计算机实现的处理，从而在所述计算机或其他可编程数据处理设备上执行的指令提供了用于实现所述流程图和/或框图块中指定的功能/行为的处理。

本发明的实施例可以通过云计算基础结构提供给终端用户。云计算通常是指提供可伸缩的计算资源作为网络上的服务。更正式地，云计算可以被定义为提供计算资源与其背后的技术基础结构(如，服务器、存储器、网络)之间的抽象的计算能力，使得能够对可配置的计算资源的共享池进行方便的、按需网络接入，所述可配置的计算资源能够在最小的管理投入或服务提供商干预的情况下被快速地提供和释放。因此，云计算允许用户访问“所述云”中的虚拟计算资源(例如，存储器、数据、应用，甚至完整的虚拟计算系统)，而不关心用于背后用于提供所述计算资源的物理系统(或这些系统的位置)。

典型地，云计算资源是在按使用量计费的基础上提供给用户的，其中，用户仅为实际使用的计算资源(如，用户消耗的存储空间的量或用户实例化的虚拟系统的数量)付费。用户能够从因特网上的任意地方并在任意时间访问驻留在云中的任意资源。在本发明的上下文中，用户可以访问云中可用的应用或相关的数据。例如，用于创建流应用的节点可以是由云服务提供方主持的虚拟机器。这么做允许用户从附接于连接到所述云的网络(如，因特网)的任意计算系统访问此信息。

图1A-1B示出了根据本发明的一个实施例的、被配置用于执行流应用的计算基础结构100。如图所示，所述计算基础结构100包括管理系统105和多个计算节点130_1-4，每个计算节点都连接到通信网络120。另外，所述管理系统105包括操作码图132和流管理器134。如下面将更详细地说明，所述操作码图132表示从一个或多个源处理元素(PE)开始直到一个或多个接收(sink)PE的流应用。此从源到接收的流在这里也通常被称为执行路径。一般地，数据元素流进流应用的源PE并被该PE处理。典型地，处理元素既从所述流接收N个元组的数据元素，也向所述流发出N个元组的数据元素(除了流终止处的接收PE)。当然，由处理元素接收到的N个元组不需要是向下游发送的N个相同元组。另外，所述处理元素可以被配置为以除N个元组之外的格式接收或发出数据(例如，所述处理元素可以交换被标记为XML文件的数据)。此外，每个处理元素可以被配置为在接收到的元组上执行任意形式的数据处理功能，包括，例如，写入到数据库表或执行其他诸如数据连接、分裂、读取等的数据库操作，以及执行其他数据分析功能或操作。

所述流管理器134可以被配置为监视在计算节点130_1-4上运行的流应用，以及改变所述操作码图134的结构。例如，所述流管理器134可以将处理元素(PE)从一个计算节点130移动到另一个，例如，用以管理所述计算基础结构100中的计算节点130的处理负荷。另外，流管理器134可以通过插入、移除、融合(fusing)、解融合(un-fusing)或者修改在所述计算节点130_1-4上运行的处理元素(或数据元组流向处理元素的内容)控制所述流应用。在本发明的一个实施例中，所述管理系统105可以维持多个操作码图132。在这样的实施例中，一个操作码图132被指派为表示一般的或默认的处理流的主操作码图，另一个操作码图可以表示替换的处理流。

图1B示出了包括十个在计算节点130_1-4上运行的处理元素(标记为PE1-PE10)的示例操作码图。虽然处理元素可以用它自己的进程ID(PID)和存储空间作为独立运行的进程而被执行，但也可以融合多个处理元素作为单个进程(具有PID和存储空间)运行。在两个(或多个)处理元素独立地运行的情况下，可以使用网络套接字(如，TCP/IP套接字)进行进程间通信。然而，当进程被融合在一起时，被融合的处理元素可以使用更快速的通信技术在处理元素(和每一个处理元素中的操作码)之间传递N个元组(或其他数据)。

如图所示，所述操作码图开始于源135(流进被标记为PE1的处理元素)且结束于接收140_1-2(从被标记为PE6和PE10的处理元素流出)。计算节点1301包括处理元素PE1、PE2和PE3。源135流进处理元素PE1，其依次发出由PE2和PE3接收的元组。例如，PE1可以分裂在元组中接收的数据元素，并将一些数据元素传递到PE2，同时将其他数据元素传递到PE3。流到PE2的数据被PE2中包含的操作码处理，且生成的元组接着被发送到计算节点130₂上的PE4。同样地，由PE4发出的数据元组流向接收PE61 40₁。类似地，从PE3流向PE5的数据元组也到达接收PE6 140₁。因此，除了作为该示例操作码图的接收之外，PE6也可以被配置为执行连接操作，组合从PE4和PE5接收到的元组。该示例操作码图还显示了从PE3流向计算节点130₃上PE7的数据元组，该计算节点本身显示了流向PE8并循环回到PE7的数据元组。从PE8发出的数据元组流向计算节点130₄上的PE9，该计算节点继而发出将由接收PE10 140₂处理的元组。

图2是根据本发明的一个实施例的、图1A-1B的计算节点130的更详细的视图。如图所示，所述计算节点130包括但不限于中央处理单元(CPU)205、网络接口215、互连220、存储器225和储存器230。所述计算节点130还可以包括用于将I/O设备212(如，键盘，显示器和鼠标设备)连接到计算节点130的I/O设备接口210。

所述CPU 205检索(retrieve)并执行存储在存储器225中的程序指令。类似地，所述CPU 205存储并检索驻留在存储器225中的应用数据。所述互连220被用于在CPU 205、I/O设备接口210、储存器230、网络接口215和存储器225之间传送程序指令和应用数据。CPU 205被包括来作为单个CPU、多个CPU、具有多个处理核的单个CPU等的代表。存储器225通常被包括来作为随机存取存储器的代表。储存器230，诸如硬盘驱动器、固态器件(SSD)或闪存储存驱动器，可以存储非易失性数据。

在此示例中，存储器225包括多个处理元素(PE)235、估计处理时间(EPT)组件(component)245和流连接数据255。每个PE 235包括一组操作码240。如上所述，每个操作码240可以提供一小块可执行代码，其被配置来处理流进处理元素(如，PE 235)的数据并将数据发送到该PE中的其他操作码240以及发送到流应用中的其他PE。这样的PE可以在同一个计算节点130上或在数据通信网络120上可访问的其他计算节点上。所述流连接数据255表示计算节点130上的PE之间的连接(如，两个分离的PE 235之间的TCP/IP套接字连接)，以及到具有所述流应用中的上游和/或下游PE的其他计算节点130的连接，也经由TCP/IP套接字(或其他进程间数据通信机制)。

缓冲的流数据260表示用于从上游处理元素(或从所述流应用的数据源)流进计算节点105的数据的存储空间。例如，缓冲的流数据可以包括等待被PE 235之一处理的数据元组。缓冲的流数据260还可以存储由PE 235执行的数据处理的结果，其将被发送到下游的处理元素。

EPT组件245可以接收要在PE 135之一上处理的数据流的进入元组。一般地，所接收到的元组可以来自源135，或可选择地，可以是系统100中另一个PE的输出。当接收到元组时，EPT组件245可以计算处理给定元组的估计处理时间。在一个实施例中，所述估计处理时间是基于给定元组中包含的一个或多个属性而计算的。在另一个实施例中，所述估计处理时间是基于收集到的关于给定元组的元数据而计算的。在再一个实施例中，所述估计处理时间是基于从使用PE 235处理的其他元组收集到的历史数据而计算的。一旦所述估计处理时间被计算，则所述EPT组件245接着可以确定所述估计处理时间是否超过预定阈值时间。如果所述估计处理时间不超过预定阈值时间，则所述EPT组件245可以将给定元组发送到适当的PE 235进行处理。

作为选择，如果所述估计处理时间超过预定阈值时间，则所述EPT组件245可以将给定元组从主执行路径移除。在本发明的一个实施例中，所述EPT组件245可以将给定元组发送到替换的执行路径进行处理。例如，假定所述主执行路径指定使用第一PE 235处理所述给定元组。如果EPT组件245确定所述估计处理时间超过预定阈值时间，则所述EPT组件245接着可以根据替换的执行路径将所述给定元组路由到第二PE 235。在本发明的另一个实施例中，当确定所述估计处理时间超过了预定阈值时间时，所述EPT组件245延迟处理给定元组。例如，所述EPT组件245可以将该元组的处理延迟到主执行路径指定的PE 235变为空闲为止。在一个替代实施例中，当确定所述估计处理时间超过预定阈值时间时，所述EPT组件245丢弃数据流的给定元组。在这样的替代实施例中，所述EPT组件245可以产生指示所述给定元组被丢弃的消息。示例地而非限制地，所述EPT组件245接着可以将所产生的消息存储在日志文件中，或可以将这样的消息传达给用户(例如，使用诸如监视器的I/O设备212)。

图3是根据本发明的一个实施例的、图1的服务器计算系统105的更详细的视图。如图所示，服务器计算系统105包括但不限于，中央处理单元(CPU)305、网络接口315、互连320、存储器325和存储器330。所述客户端系统130还可以包括将I/O设备312(如，键盘、显示器和鼠标设备)连接到服务器计算系统105的I/O设备接口310。

类似于图2的CPU 205，CPU 305被配置为检索和执行存储在存储器325和储存器330中的程序指令。类似地，所述CPU 305被配置为存储和检索驻留在存储器325和储存器330中的应用数据。所述互连320被配置为在CPU305、I/O设备接口310、储存器330、网络接口305和存储器325之间移动诸如程序指令和应用数据的数据。类似CPU 205，CPU 305被包括来作为单个CPU、多个CPU、具有多个处理核的单个CPU等的代表。存储器325通常被包括来作为随机存取存储器的代表。网络接口315被配置为经由通信网络120传送数据。虽然被显示为单个单元，但储存器330可以是固定的和/或可移除的存储设备的组合，诸如固定的硬盘驱动器、可移除的存储卡、光学存储器、SSD或闪存设备、网络附接存储器(NAS)、或到存储区域网络(SAN)设备的连接。

如图所示，存储器325存储流管理器132。另外，所述存储器330包括主操作码图335和替换的操作码图340。所述流管理器132一般可以根据主操作码图335将作为数据流的一部分接收的元组路由到PE 235进行处理。如上所述，在本发明的一个实施例中，如果EPT组件245确定所述估计处理时间超过预定阈值时间，则所述EPT组件245接着可以根据替换的执行路径将给定元组路由到第二PE 235。在这样的情形中，所述替换的执行路径可以由替换的操作表240指定。此外，所述EPT组件245可以被配置为通知流管理器132根据替换的操作码表340将针对该数据流接收的所有随后接收到的元组路由到PE 235。

图4说明了根据本发明的一个实施例的、流应用中的计算节点的示例。如图所示，计算节点130₂包括三个处理元素235₂、235₃和235₄(被标记为PE1-PE3)。在所描述的示例中，处理元素235₂接收N元组数据流并将N个元组发送到计算节点130₃上的处理元素235₃和235₄(分别被标记为PE2和PE3)。处理元素PE2和PE3依次将N个元组发送到计算节点130₃上的处理元素235₅。在这个示例中，所述PE 135(被标记为PE1)接收包括属性<姓名，部门，薪金，性别>的元组。PE1取得该N个元组并基于PE1接收到的元组中的性别值产生一组发送到PE2的元组以及另一组发送到PE 3的元组。依次地，PE2和PE3为从PE1接收到的每个元组执行数据库写操作并产生发送到PE4的属性元组<姓名，部门，薪金>。一旦被接收，PE4就访问第三方网站服务并产生进一步向流应用的下游发送的元组。

继续这个示例，所描述的流应用可以是这样：所接收到的元组需要尽可能接近实时地被处理。如此，所述流应用可以以流水线的方式处理元组。如这里所使用的，流水线处理是如下的任意处理：第一处理元素处理的结果被下一个处理元素处理，由所述下一个处理元素处理的结果被随后的处理元素处理、并依次类推。因此，使用图4中描绘的示例，所述处理被认为是流水线式的是因为由PE1处理的结果被提供给PE2或PE3用于处理，且PE2和PE3的结果被提供给PE4进行进一步的处理。

因为所述处理的流水线性质，需要更长时间处理的特定数据元组不仅延迟了特定元组的处理，还延迟了所有随后接收到的元组的处理。例如，如果平均每5毫秒接收一个元组，且示例的流应用平均用4毫秒处理一个元组，则所述流应用将能够在不积压未处理的元组的情况下处理接收到的元组。然而，进一步假设接收到的特殊元组花费5分钟处理。这是可能发生的，因为，例如，该特殊元组可能包含需要访问第三方服务而进行处理的特定属性。例如，如果“部门(DEPT.)”属性包含值“工程部(ENGINEERING)”，则所述流应用可能被配置为访问工程部特定的第三方薪水单服务。如果所述流应用和所述第三方服务之间的连接当前遭遇问题，则所述特殊元组的处理可能花费比正常完成长得多的时间(例如，5分钟对4毫秒)。如此，因为示例中的特殊元组的处理要求比其他元组的平均处理时间长得多的处理时间，所以所述流应用将积压在5分钟的处理时间内接收到的所有其他元组，所有这些元组的处理都被延迟。此外，所述积压(相应地，延迟)将随着其他包含用于“部门”属性的值“工程部”的元组的接收而继续加重。

本发明的实施例可以基于属性概要分析移除被预测为需要花费过分长的时间量进行处理的特定运行时数据。根据本发明的实施例，所述EPT组件245可以在元组被处理之前概要分析接收到的每个元组。所述概要分析可以考虑包含在该元组中的属性、所述属性的值、关于所述属性的元数据、从其他元组的处理收集到的历史数据，等等。一旦产生了简档(profile)，EPT组件245可以为接收到的元组计算估计处理时间。所述EPT组件245接着可以确定所述估计处理时间是否小于或等于预定阈值时间，且如果是，则可以根据主执行路径(即，用于接收到的元组的标准处理路径)将接收到的元组发送到适当的处理元素进行处理。然而，如果所述EPT组件245确定估计处理时间超过预定处理时间，则所述EPT组件245可以避免使用该处理元素处理所接收到的元组。作为替代，可以使用替换的执行路径接着处理所接收到的元组。在本发明的另一个实施例中，所接收到的元组的处理可以被延迟到以后的时间(例如，当处理元素变得空闲的时候，或者当所述连接不再遭遇问题的时候)。在又一个实施例中，所接收到的元组可以被简单地丢弃并不处理。

通过避免当所述估计处理时间超过阈值时间时使用主执行路径处理接收到的元组，本发明的实施例帮助避免了特定元组在流应用中制造积压的情况。此外，本发明的实施例可以被配置为延迟或重新路由所述特定元组的处理，使得元组中包含的数据对于流应用不被丢失。这些优势尤其与接收到的数据需要接收到时就被处理且尽可能地实时处理的流应用相关。

另外，本发明的实施例可以被配置为基于一个或多个因素计算阈值时间量而不是使用预定的阈值时间量。在本发明的一个实施例中，所述EPT组件245基于系统的工作负荷(或期望的工作负荷)计算阈值时间量。例如，在高峰处理时间期间，所述EPT组件245可以将阈值时间量设置为相对较短的时间量，因为系统资源在此时间内非常紧缺。作为另一个示例，在当所述系统资源空闲或小于最大容量操作时的非高峰处理时间期间，所述EPT组件245可以将阈值时间量设置为相对较长的时间量。有利地，以这种方式，本发明的实施例可以进行一种负载平衡，因为当系统在繁重的工作负荷下操作时冗长的查询(即，在处理时间方面)可能是受限制的。

图5是说明了根据本发明的一个实施例的预测性移除运行时数据的方法的流程图。如图所示，方法500开始于步骤520，其中所述EPT组件245接收元组以处理。在一个实施例中，所述元组可以被直接发送到EPT组件245。在一个替换的实施例中，所述EPT组件245可以在元组被处理元素235接收之前拦截被发送到处理元素235的元组。

一旦所述EPT组件245接收到所述元组，EPT组件245产生用于接收到的元组的简档(步骤522)。所述简档一般可以包括与接收到的元组相关的信息。例如，所述简档可以包括关于哪些属性被包含在接收到的元组中的信息，还可以包括关于那些属性的值的信息。另外，所述简档可以包含关于接收到的元组的元数据。示例性的元数据可以包括但不限于：包含在所述元组中的属性的数量、包含在所述元组中的属性的大小、包含在所述元组中的属性的类型以及所述元组所属的分组。所述简档可以进一步包括针对先前处理过的元组所收集的信息。

一旦所述EPT组件245产生所接收到的元组的简档，则EPT组件245计算处理所接收到的元组的估计处理时间(步骤524)。所述EPT组件245可以基于为所接收到的元组所产生的简挡而进行该确定。例如，因为所接收到的元组包含用于“部门”属性的值“市场部(MARKETING)”，所以EPT组件245可以确定所述元组将花费约15毫秒进行处理。

一旦所述估计处理时间被计算，则所述EPT组件245将估计的处理时间与预定阈值时间量进行比较(步骤526)。如果所述EPT组件245确定估计的处理时间大于阈值时间，则EPT组件丢弃所接收到的元组(步骤530)。所述EPT组件245接着将关于被丢弃的元组的信息记录为日志(步骤532)。在一个实施例中，所述EPT组件也可以将所接收到的元组中包含的属性的值记录为日志，使得所述元组可以在以后的时间被处理。在替换的实施例中，所述EPT组件245可以替代地将所接收到的元组路由到替换的执行路径进行处理。另外，所述EPT组件245还可以被配置为将所有随后从所接收到的元组所属的数据流接收到的元组路由到替换的执行路径。

替代地，如果所述EPT组件245确定估计的处理时间不大于阈值时间量，则所述EPT组件245可以根据主执行路径将接收到的元组发送到适当的处理元素，接着在主执行路径中处理所述元组(步骤528)。一旦所述元组被处理，或一旦所述EPT组件245已经将关于被丢弃的元组的信息记录为日志，则所述EPT组件245确定是否存在更多的元组要处理(步骤534)。如果所述EPT组件245确定还存在更多的元组要处理，则所述方法再次从步骤520开始，其中，EPT组件245接收下一个要处理的元组。替代地，如果所述EPT组件245确定不存在更多的元组要处理，则方法500结束。

使用方法500的一个优势是，所述流应用可以避免处理使用主执行路径可能引起实质性(substantial)延迟的元组。如此，通过仅仅处理那些被估计能在阈值时间量内完成处理的元组，所述方法500帮助避免处理任何可能引起“瓶颈”效应并在流应用中制造要被处理的元组的积压的元组。此外，本发明的实施例可以被配置为延迟其处理时间很有可能超过阈值时间量的元组的处理，或者替代地，可以使用替换的执行路径处理这些元组。如此，这些元组中包含的数据就不会丢失，同时，主执行路径也不会由于处理它们而被延迟。

图6是说明了根据本发明的一个实施例的概要分析元组的方法的流程图。如图所示，方法600开始于步骤620，其中所述EPT组件245计算关于接收到的元组的元数据。如上所述，示例的元数据可以包括所述元组中包含的属性的数量、所述元组中包含的属性的大小、所述元组中包含的属性的类型以及所述元组所属的分组。当然，上述示例仅仅是为了说明性的目的，本领域的普通技术人员应该意识到，也可以收集关于接收到的元组的其他元数据。

一旦计算了元数据，所述EPT组件245就分析所接收到的元组中包含的属性的值(步骤622)。例如，所述EPT组件245可以确定所接收到的元组包含用于“部门”属性的值“工程部”。所述EPT组件245接着检索与所接收到的元组相关的历史数据(步骤624)。例如，如果所接收到的元组包含具有值“工程部”的“部门”属性，则所述EPT组件245可以检索与处理其他包含用于“部门”属性的值“工程部”的元组相关的历史数据。一旦检索到所述历史数据，EPT组件245则创建用于检索到的元组的简档(步骤626)。

通过产生用于接收到的元组的简档，所述EPT组件245接着可以使用所产生的简档估计所述元组的处理时间。例如，因为特定的元组包含用于“部门”属性的值“工程部”，所以所述EPT组件245可以估计所述元组将花费50毫秒的时间处理。作为另一个示例，所述简档可能包含关于其他相关的元组的处理的历史信息。例如，简档可以包含指示分别用了10毫秒、20毫秒和30毫秒的处理时间处理了包含用于“部门”属性的值“工程部”的三个元组的信息。如此，所述EPT组件245可以确定包含用于“部门”属性的值“工程部”的元组的估计处理时间将是所述三个之前记录的时间的平均值，并因而是20毫秒。通过考虑这些和其他因素，所述EPT组件245可以产生用于所接收到的元组的全面简档，其接着可以被用于计算所述元组的准确的估计处理时间。此外，虽然以上示例关注于分析元组中的单个属性，本发明的实施例在计算所述元组的估计处理时间时当然还可以考虑接收到的元组中的多个或所有属性，以及其他相关信息。

附图中的流程图和框图说明了根据本发明的各个实施例的系统、方法和计算机程序产品的可能实施方式的体系结构、功能和操作。就这一点而言，所述流程图或框图中的每个块可以表示模块、部分或代码段，其包括一个或多个用于实现指定的逻辑功能的可执行指令。还应该注意到，在一些替换的实施方式中，所述块中标注的功能可以不按照图中标注的顺序出现。例如，顺序显示的两个块可以实际上被大致并发地执行，或所述块有时可以以相反的顺序被执行，取决于涉及的功能。还应该注意到，所述框图和/或流程图中的每个块，以及所述框图和/或流程图中的块的组合，可以通过执行指定功能或行为的特殊用途的硬件系统或者特殊用途的硬件和计算机指令的组合来实现。

虽然上文针对本发明的实施例，但本发明的其他或进一步的实施例可以在不脱离其基本范围的情况下被构思出，且其范围是由权利要求所确定。

Claims

1.一种用于处理数据流的计算机实现的方法，包括：

接收要处理的数据流的第一元组，其中所述第一元组包括多个属性；

通过一个或多个计算机处理器的操作，至少部分地基于所述第一元组中的多个属性中的一个或多个属性，计算用于处理所述数据流的所接收到的第一元组的估计处理时间；

当确定所计算的估计处理时间不超过预定阈值时间时，使用第一执行路径以第一方式来处理所述数据流的第一元组；以及

当确定所计算的估计处理时间超过所述预定阈值时间时，以第二方式处理所述数据流的所接收到的第一元组，相对于以所述第一方式处理所述第一元组，所述第二方式降低所述第一执行路径中的处理延迟。

2.如权利要求1所述的方法，还包括：

当确定所计算的估计处理时间超过预定阈值时间时，丢弃所述数据流的所接收到的第一元组。

3.如权利要求1所述的方法，还包括：

当确定所计算的估计处理时间超过预定的阈值时间时：

存储所述数据流的所接收到的第一元组，其中，将在能够获得空闲资源时处理所存储的第一元组；并且

当确定在所述第一执行路径中至少能够获得一个空闲资源时，使用所述第一执行路径处理所存储的第一元组。

4.如权利要求1所述的方法，还包括：

当确定所计算的估计处理时间超过预定阈值时间时，使用替换的执行路径处理所述数据流的所接收到的第一元组。

5.如权利要求4所述的方法，还包括：

当确定所计算的估计执行时间超过预定阈值时间时，使用所述替换的执行路径处理所述数据流的所有后续元组。

6.如权利要求1所述的方法，其中使用所述第一执行路径处理所述数据流的所接收到的第一元组还包括：

计算使用所述第一执行路径来处理所述数据流的所接收到的第一元组的实际执行时间；以及

将所述实际执行时间和所处理的第一元组的标识符存储到历史数据。

7.如权利要求1所述的方法，其中计算用于处理所述数据流的所接收到的第一元组的估计处理时间还基于从处理其他元组而收集到的历史数据。

8.一种系统，包括：

计算机处理器，以及

包含程序的存储器，当在所述计算机处理器上执行所述程序时，所述程序执行用于处理数据流的操作，所述操作包括：

至少部分地基于所述第一元组中的多个属性中的一个或多个属性，计算用于处理所述数据流的所接收到的第一元组的估计处理时间；

当确定所计算的估计处理时间不超过预定阈值时间时，使用第一执行路径以第一方式来处理所述数据流的所接收到的第一元组；以及

9.如权利要求8所述的系统，所述操作还包括：

10.如权利要求8所述的系统，所述操作还包括：

当确定所计算的估计处理时间超过预定阈值时间时：

11.如权利要求8所述的系统，所述操作还包括：

12.如权利要求8所述的系统，其中所述系统还包括：

在一个或多个计算节点上操作的多个处理元素；

其中，所述多个处理元素中的处理元素被配置为根据指定的执行路径以流水线的方式处理所述数据流的元组，以及

其中，所述多个处理元素中的每个处理元素被配置为通过对元组执行一个或多个操作来处理数据流的元组。

13.如权利要求8所述的系统，其中使用所述第一执行路径处理所述数据流的所接收到的第一元组还包括：

计算使用所述第一执行路径处理所述数据流的所接收到的第一元组的实际执行时间；以及

14.如权利要求8所述的系统，其中，计算用于处理所述数据流的所接收到的第一元组的估计处理时间还基于从处理其他元组而收集到的历史数据。

15.一种用于处理数据流的计算机实现的方法，包括：

通过一个或多个计算机处理器的操作，至少部分地基于与所接收到的第一元组相关联的元数据，计算用于处理所述数据流的所接收到的第一元组的估计处理时间；

当确定所计算的估计处理时间超过预定阈值时间时，以第二方式处理所述数据流的第一元组，所述第二方式最小化在所述第一执行路径中处理元素的任何延迟。

16.如权利要求15所述的方法，还包括：收集描述所述数据流的所接收到的第一元组的元数据。

17.如权利要求15所述的方法，其中所述元数据包括以下中的至少一个：(i)所述多个属性中的属性的数目；(ii)所述多个属性中的一个或多个属性的长度；以及(iii)所述多个属性中的一个或多个属性的类型。