Distributed event processing

• US 10,387,225 B2
• Filed: 09/30/2015
• Issued: 08/20/2019
• Est. Priority Date: 08/06/2015
• Status: Active Grant

• Current Assignee: Hewlett Packard Enterprise Development LP (Hewlett-Packard Enterprise Company)
• Original Assignee: Hewlett Packard Enterprise Development LP (Hewlett-Packard Enterprise Company)
• Inventors: Guigui, Alain, Verney, Remi, Roger, Denis
• Primary Examiner(s): Dorais, Craig C
• Application Number:
  US15/750,090
• Publication Number:
  US 20180225158A1
• Time in Patent Office: 1,420 Days
• Field of Search: None
• CPC Class Codes: G06F 2209/505 Clust
• CPC Class Codes: G06F 9/5077 Logical partitioning of resources; Management or configuration of virtualized resources specific details on emulation or internal functioning of virtual machines G06F9/455
• CPC Class Codes: G06F 9/542 Event management; Broadcasting; Multicasting; Notifications

Abstract

Certain examples described herein provide a data processing system and method adapted for event processing. These examples provide for distribution of data processing operations between server computing devices. In one case, a plurality of processing stages are implemented using computing instances on the server computing devices. In this case, the computing instances are assigned to the server computing devices in order to perform at least one data processing operation in parallel. Certain examples described herein then provide for the distribution of data between computing instances such that parallelism is maintained for data processing operations. In certain cases, a composite key is used. In this case, a composite key value is computed for a set of data fields associated with a data item to be processed. This key value is computed based on a data processing operation to be performed. The key value is used to route the data item to an associated computing instance implementing the data processing operation.

Claims

1. A data processing system comprising:

a complex event processing engine to distribute real-time data processing operations between server computing devices, the complex-event processing engine being configured to;

load configuration data defining processing stages in a processing pipeline, the processing pipeline being configured to process at least one data record stream;

instruct a distributed assignment of computing instances across the server computing devices, each server computing device loading a computing instance into memory for execution by at least one processor, the computing instances implementing each processing stage in parallel; and

instruct data distribution between the computing instances implementing each processing stage,

wherein, for a stateful processing stage, data distribution is performed based on a composite key computed according to a data processing operation performed by the stateful processing stage such that input data records from a previous processing stage having a common composite key value are sent to a common computing instance for the stateful processing stage.

2. The data processing system of claim 1, comprising:

a graphical user interface (220) to receive commands from a client device relating to an ordering of the processing stages In the processing pipeline and to store the configuration data relating to the processing pipeline following the ordering.

3. The data processing system of claim 1, wherein each processing stage comprises one of:

a data loader to read data from at least one file or stream;

an event processor to perform a defined data processing operation; and

a data writer to save data to at least one file or stream.

4. The data processing system of claim 1, wherein the complex event processing engine and the server computing devices are implemented as virtualized network functions.

5. The data processing system of claim 1, wherein the complex event processing engine comprises:

a routing component to route a data record from a computing instance of a first processing stage to a computing instance of a second processing stage,

wherein, responsive to the second processing stage being stateless, the routing component is configured to select the computing instance of the second processing stage based on a load balancing function, and

wherein, responsive to the second processing stage being stateful, the routing component is configured to;

compute a value for the composite key based on a data processing operation performed by the second processing stage;

identify a computing instance of the second processing stage that is configured to process data records having the computed value for the composite key; and

route the data record to the identified computing instance.

6. The data processing system of claim 1, wherein at least one processing stage comprises one of the following data processing operations:

an aggregation operation;

a correlation operation;

a geolocation tracking operation; and

a pattern detection operation,

wherein for the aggregation operation the composite key comprises a hash of a set of aggregated data field values,

wherein for the correlation operation the composite key comprises a hash of a set of correlated data field values,

wherein for the geolocation tracking operation the composite key comprises a hash of a set of data field values associated with a moving entity, and

wherein for the pattern detection operation the composite key comprises a hash of a set of data field values in which a pattern is to be detected.

7. The data processing system of claim 1, comprising:

a cluster of server computing devices, each server computing device comprising memory and at least one processor, the cluster of server computing devices being communicatively coupled to the complex event processing engine and configured to continuously implement a plurality of computing instances in said memory.

8. A distributed data processing method comprising:

obtaining a data item output by a first computing instance, the first computing instance forming part of a plurality of computing instances configured to implement in parallel a first processing stage in a plurality of interconnected processing stages, the data item being associated with an event occurring at a given time;

determining, from data defining the plurality of interconnected processing stages, a second processing stage in the plurality of interconnected processing stages configured to receive data from the first processing stage;

determining whether the second processing stage is configured to process each data item independently of other data items; and

responsive to a determination that the second processing stage is not configured to process each data item independently of other data items;

computing a composite key value from fields associated with the obtained data item;

determining a second computing instance corresponding to the computed composite key value from a plurality of computing instances implementing the second processing stage in parallel; and

sending the data item output by the first computing instance to the determined second computing instance.

9. The method of claim 8, comprising:

responsive to a determination that the second processing stage is configured to process each data item independently of other data items, determining a second computing instance from a plurality of computing instances implementing the second processing stage in parallel according to a load balancing function.

10. The method of claim 8, comprising:

obtaining a data item output by the second computing instance;

determining, from data defining the plurality of interconnected processing stages, a third processing stage in the plurality of interconnected processing stages configured to receive data from the second processing stage;

determining whether the third processing stage is configured to process each data item independently of other data items; and

responsive to a determination that the third processing stage is configured to process each data item independently of other data items;

determining a third computing instance from a plurality of computing instances implementing, in parallel, the third processing stage according to a load balancing function, or

responsive to a determination that the third processing stage is not configured to process each data item independently of other data items;

computing a composite key value from fields associated with the data item output by the second computing instance;

determining a third computing instance corresponding to the computed composite key value from the plurality of computing instances implementing the third processing stage; and

sending the data item output by the second computing instance to the determined third computing instance.

11. The method of claim 8, wherein:

the first computing instance comprises machine readable instructions that are stored in memory of a first server computing device and that are executed by at least one processor of the first server computing device;

the second computing instance comprises machine readable instructions that are stored in memory of a second server computing device and that are executed by at least one processor of the second server computing device, and

the first and second server computing devices are communicatively coupled.

12. The method of claim 8, comprising:

receiving, by the second computing instance, the data item; and

aggregating or correlating, by the second computing instance, fields associated with the data item,

wherein the composite key value is calculated as a function of a respective one of the aggregated or correlated fields.

13. The method of claim 8, comprising, prior to obtaining the data item output by the first computing instance:

obtaining input data items from at least one real-time event stream and passing said data items to the plurality of computing instances configured to implement in the first processing stage; and

outputting processed data items from the plurality of computing instances implementing the first processing stage.

14. The method of claim 8, comprising:

defining at least the first and second processing stages, and at least one interconnection between said stages, using a graphical user interface; and

storing data defining the first and second processing stages and the at least one interconnection.

15. A non-transitory machine readable medium comprising instructions which, when loaded into memory and executed by at least one processor of a complex event processing engine, cause the at least one processor to:

retrieve a configuration file defining an event processing pipeline, the event processing pipeline comprising a plurality of event processing operations and a plurality of connections coupling said operations indicating a logical sequence for the event processing pipeline;

initialize a plurality of computing instances for each event processing operation across a cluster of server computing devices, the plurality of computing instances being configured to perform the event processing operation in parallel;

obtain events from at least one real-time event stream, each event comprising at !east one data field; and

distribute the obtained events between the plurality of computing instances, wherein each computing instance is configured to receive input data associated with an event and to output data following performance of the event processing operation, and

wherein, for an event processing operation based on a relationship between events the instructions cause the at least one processor to;

route events from computing instances of a previous event processing operation to computing instances of a subsequent event processing operation based on a compound index defined based on the subsequent event processing operation, such that events with the same compound index are sent to the same computing instance.