Distributed complex event processing

• US 9,992,269 B1
• Filed: 02/25/2014
• Issued: 06/05/2018
• Est. Priority Date: 02/25/2013
• Status: Active Grant

• Current Assignee: Emc IP Holding Company LLC (Dell Technologies Inc.)
• Original Assignee: Emc IP Holding Company LLC (Dell Technologies Inc.)
• Inventors: Odom, Jeffrey M., Fikes, Michael P., Swift, John
• Primary Examiner(s): Serrao, Ranodhi
• Application Number:
  US14/189,326
• Time in Patent Office: 1,561 Days
• Field of Search: 709201, 709203, 709224, 707769
• CPC Class Codes: G06F 16/24568 Data stream processing; Continuous queries
• CPC Class Codes: G06F 9/505 considering the load
• CPC Class Codes: H04L 67/10 in which an application is distributed across nodes in the network software deployment G06F8/60; multiprogramming arrangements G06F9/46

Abstract

A complex event query specification may be constructed that a complex event processing engine uses to analyze an event stream for an occurrence of a complex event. Event data stored in a distributed file system are mapped to respective instances of the complex event processing engine in respective distributed event streams. The distributed event streams are analyzed by the independently executing complex event processing engines in accordance with the complex event query specification. The occurrence of the complex event in any of the distributed event streams is indicated at the output of the complex event processing engines.

Claims

1. A method comprising:

constructing at least one complex event query specification comprising one or more grouping criteria by which event data is grouped into distributed event streams, one or more filtering criteria by which individual events relevant to a given complex event are specified and one or more query statements specifying event stream processing to be performed by complex event processing engines;

establishing a time interval defining an event stream continuation interval for mapping the distributed event streams to complex event processing engines;

analyzing said at least one complex event query specification to determine a given number of different complex event processing engines and corresponding distributed event streams based on the grouping criteria in said at least one complex event query specification;

instantiating the given number of complex event processing engines;

activating said at least one complex event query specification in the complex event processing engines; and

responsive to determining that a new event stream continuation interval has arrived based on the established time interval;

identifying new event data stored in a distributed file system relevant to said at least one complex event query specification, wherein relevance is determined based on the one or more filtering criteria in said at least one complex event query specification; and

mapping the new event data to respective ones of the complex event processing engines via the distributed event streams, wherein the complex event processing engines evaluates the one or more query statements of said at least one complex event query specification for the event data in the distributed event streams;

wherein the method is performed by at least one processing device comprising a processor coupled to a memory, the at least one processing device comprising one or more master processing nodes and one or more slave processing nodes, the complex event processing engines being instantiated on respective ones of the slave processing nodes, the slave processing nodes executing under control of the master processing nodes;

wherein the master processing nodes are configured to implement an application interface for constructing the at least one complex event processing query, to coordinate use of compute and storage resources by the complex event processing engines, to track one or more complex event processing jobs performed by the complex event processing engines, and to control and record transactions of the complex event processing jobs in a distributed file system;

wherein a given one of the slave processing nodes is configured to monitor one or more of the complex event processing jobs, to report job progress for the one or more complex event processing jobs to the master processing nodes, to perform file system transactions for one or more of the complex event processing jobs in the distributed file system, and to provide event data from the distributed file system to a given one of the complex event processing engines;

wherein said at least one complex event query specification comprises two or more complex event query specifications;

wherein a first one of the complex event processing engines and at least a second one of the complex event processing engines are activated with a same one of the two or more complex event query specifications; and

wherein a third one of the complex event processing engines and at least a fourth one of the complex event processing engines are activated with different ones of the two or more complex event query specifications.

2. The method of claim 1 wherein the one or more query statements of said at least one complex event query specification comprise one or more complex event query language statements that together define a complex event in terms of conditions on and relationships between individual events represented in one or more of the distributed event streams.

3. The method of claim 2 wherein the complex event query language utilizes an object-oriented programming paradigm where events are defined as object classes.

4. The method of claim 1 wherein the master processing nodes and the slave processing nodes are part of a data warehouse comprising a plurality of equipment racks, each equipment rack comprising interconnect equipment and at least one of a process-centric appliance and a storage-centric appliance, the process-centric appliances comprising greater processing capabilities than the storage-centric appliances and the storage-centric appliances comprising larger storage capacity than the process-centric appliances.

5. The method of claim 4 wherein the master processing nodes are implemented by one or more of the process-centric appliances and the slave processing nodes are implemented by one or more of the storage-centric appliances.

6. The method of claim 1 wherein identifying the new event data stored in the distributed file system relevant to said at least one complex event query specification comprises:

obtaining an event record from the distributed file system; and

determining whether the obtained event record meets filter conditions specified in the one or more filtering criteria of said at least one complex event query specification.

7. The method of claim 6 wherein mapping the new event data comprises, responsive to determining that the obtained event record meets filter conditions specified in the one or more filtering criteria of said at least one complex event query specification:

hashing one or more attributes of the obtained record to obtain a hash value, the one or more attributes being selected based on the one or more grouping criteria of said at least one complex event query specification;

matching the hash value with a given one of the complex event processing engines; and

providing the obtained record to the given complex event processing engine via its associated distributed event stream.

8. The method of claim 7 wherein providing the obtained record to the given complex event processing engine comprises:

obtaining an address of the given complex event processing engine associated with the hash value;

constructing a hypertext transfer protocol (HTTP) request targeting the address of the given complex event processing engine, wherein a body of the HTTP request comprise the obtained event record; and

sending the HTTP request to the given complex event processing engine on a transmission control protocol (TCP) port designated by the given complex event processing engine for event stream traffic of its associated distributed event stream.

9. The method of claim 8 wherein the given complex event processing engine is configured:

to obtain the HTTP request via an input adapter listening on the TCP port designated for event stream traffic;

to retrieve the obtained event record; and

to reformat the obtained event record into a format utilized by the given complex event processing engine.

10. The method of claim 1 further comprising obtaining analysis results from the complex event processing engines and storing the analysis results in the distributed file system, the analysis results comprising detection of one or more instances of the given complex event.

11. The method of claim 1 wherein mapping the new event data as the distributed event streams comprises partitioning the new event data such that different subsets of the new event data are provided to two or more different complex event processing engines.

12. The method of claim 1 wherein the distributed file system comprises a Hadoop distributed file system (HDFS) and at least one of the complex event processing engines comprises an Esper complex event processing engine.

13. An apparatus comprising:

at least one processing device comprising a processor coupled to a memory;

the at least one processing device being configured;

to construct at least one complex event query specification comprising one or more grouping criteria by which event data is grouped into distributed event streams, one or more filtering criteria by which individual events relevant to a given complex event are specified and one or more query statements specifying event stream processing to be performed by complex event processing engines;

to establish a time interval defining an event stream continuation interval for mapping the distributed event streams to complex event processing engines;

to analyze said at least one complex event query specification to determine a given number of different complex event processing engines and corresponding distributed event streams based on the grouping criteria in said at least one complex event query specification;

to instantiate the given number of complex event processing engines;

to activate said at least one complex event query specification in the complex event processing engines; and

responsive to determining that a new event stream continuation interval has arrived based on the established time interval;

to identify new event data stored in a distributed file system relevant to said at least one complex event query specification, wherein relevance is determined based on the one or more filtering criteria in said at least one complex event query specification; and

to map the new event data to respective ones of the complex event processing engines via the distributed event streams, wherein the complex event processing engines evaluates the one or more query statements of said at least one complex event query specification for the event data in the distributed event streams;

wherein the at least one processing device comprises one or more master processing nodes and one or more slave processing nodes, the complex event processing engines being instantiated on respective ones of the slave processing nodes, the slave processing nodes executing under control of the master processing nodes;

wherein the master processing nodes are configured to implement an application interface for constructing the at least one complex event processing query, to coordinate use of compute and storage resources by the complex event processing engines, to track one or more complex event processing jobs performed by the complex event processing engines, and to control and record transactions of the complex event processing jobs in a distributed file system;

wherein a given one of the slave processing nodes is configured to monitor one or more of the complex event processing jobs, to report job progress for the one or more complex event processing jobs to the master processing nodes, to perform file system transactions for one or more of the complex event processing jobs in the distributed file system, and to provide event data from the distributed file system to a given one of the complex event processing engines;

wherein said at least one complex event query specification comprises two or more complex event query specifications;

wherein a first one of the complex event processing engines and at least a second one of the complex event processing engines are activated with a same one of the two or more complex event query specifications; and

wherein a third one of the complex event processing engines and at least a fourth one of the complex event processing engines are activated with different ones of the two or more complex event query specifications.

14. The apparatus of claim 13 wherein the one or more query statements of said at least one complex event query specification comprise one or more complex event query language statements that together define a complex event in terms of conditions on and relationships between individual events represented in one or more of the distributed event streams, and wherein the complex event query language utilizes an object-oriented programming paradigm where events are defined as object classes.

15. The apparatus of claim 13 wherein the master processing nodes and the slave processing nodes are part of a data warehouse comprising a plurality of equipment racks, each equipment rack comprising interconnect equipment and at least one of a process-centric appliance and a storage-centric appliance, the process-centric appliances comprising greater processing capabilities than the storage-centric appliances and the storage-centric appliances comprising larger storage capacity than the process-centric appliances, the master processing nodes being implemented by at least one of the process-centric appliances and the slave processing nodes being implemented by one or more of the storage-centric appliances.

16. A computer program product comprising a non-transitory processor-readable storage medium having stored therein program code of one or more software programs, wherein the program code when executed by at least one processing device cause the at least one processing device:

to construct at least one complex event query specification comprising one or more grouping criteria by which event data is grouped into distributed event streams, one or more filtering criteria by which individual events relevant to a given complex event are specified and one or more query statements specifying event stream processing to be performed by complex event processing engines;

to establish a time interval defining an event stream continuation interval for mapping the distributed event streams to complex event processing engines;

to analyze said at least one complex event query specification to determine a given number of different complex event processing engines and corresponding distributed event streams based on the grouping criteria in said at least one complex event query specification;

to instantiate the given number of complex event processing engines;

to activate said at least one complex event query specification in the complex event processing engines; and

responsive to determining that a new event stream continuation interval has arrived based on the established time interval;

to identify new event data stored in a distributed file system relevant to said at least one complex event query specification, wherein relevance is determined based on the one or more filtering criteria in said at least one complex event query specification; and

to map the new event data to respective ones of the complex event processing engines via the distributed event streams, wherein the complex event processing engines evaluates the one or more query statements of said at least one complex event query specification for the event data in the distributed event streams;

wherein the at least one processing device comprises one or more master processing nodes and one or more slave processing nodes, the complex event processing engines being instantiated on respective ones of the slave processing nodes, the slave processing nodes executing under control of the master processing nodes;

wherein the master processing nodes are configured to implement an application interface for constructing the at least one complex event processing query, to coordinate use of compute and storage resources by the complex event processing engines, to track one or more complex event processing jobs performed by the complex event processing engines, and to control and record transactions of the complex event processing jobs in a distributed file system; and

wherein a given one of the slave processing nodes is configured to monitor one or more of the complex event processing jobs, to report job progress for the one or more complex event processing jobs to the master processing nodes, to perform file system transactions for one or more of the complex event processing jobs in the distributed file system, and to provide event data from the distributed file system to a given one of the complex event processing engines;

wherein said at least one complex event query specification comprises two or more complex event query specifications;

wherein a first one of the complex event processing engines and at least a second one of the complex event processing engines are activated with a same one of the two or more complex event query specifications; and

wherein a third one of the complex event processing engines and at least a fourth one of the complex event processing engines are activated with different ones of the two or more complex event query specifications.

17. The computer program product of claim 16 wherein the one or more query statements of said at least one complex event query specification comprise one or more complex event query language statements that together define a complex event in terms of conditions on and relationships between individual events represented in one or more of the distributed event streams, and wherein the complex event query language utilizes an object-oriented programming paradigm where events are defined as object classes.

18. The computer program product of claim 16 wherein the master processing nodes and the slave processing nodes are part of a data warehouse comprising a plurality of equipment racks, each equipment rack comprising interconnect equipment and at least one of a process-centric appliance and a storage-centric appliance, the process-centric appliances comprising greater processing capabilities than the storage-centric appliances and the storage-centric appliances comprising larger storage capacity than the process-centric appliances, the master processing nodes being implemented by at least one of the process-centric appliances and the slave processing nodes being implemented by one or more of the storage-centric appliances.