Complex event processor for historic/live/replayed data

Patent metadata

Field	Value
Publication number	US-9632846-B2
Application number	US-201514677633-A
Country	US
Kind code	B2
Filing date	Apr 2, 2015
Priority date	Apr 2, 2015
Publication date	Apr 25, 2017
Grant date	Apr 25, 2017

Abstract

A complex event processor is described which has a communications interface configured to retrieve event data by pulling it from one or more sources and to receive at least one live event stream pushed to the interface. An event processing pipeline connected to the communications interface comprises a plurality of operator nodes connected between the communications interface and a combiner node which is a node configured to combine event data from the source and from the live event stream. The communications interface is configured to retrieve events from the source and to push the events retrieved from the source along the event processing pipeline towards the combiner node. The communications interface is configured to retrieve and push the retrieved events only in response to request messages passed in an upstream direction from the combiner node to the communications interface along the pipeline.

First claim

The invention claimed is:

1. A system comprising:
at least one processor adapted to execute computer executable instructions associated with nodes of an event processing pipeline;
a communications interface communicatively coupled to the at least one processor and configured to:
pull first event data from at least one source in response to a request message; and
receive second event data pushed to the communications interface, the second event data associated with at least one live/replayed event stream; and

at least one computer-readable medium communicatively coupled to the at least one processor and nodes of the event processing pipeline stored on the computer-readable medium and executable by the processor, the nodes comprising:
a first group of operator node(s) connected between the communications interface and a combiner node, the first group of operator node(s) configured to provide third event data based at least in part on the first event data;
a second group of operator node(s) connected between the communications interface and the combiner node, the second group of operator node(s) configured to provide fourth event data based at least in part on the second event data; and
the combiner node configured to:
determine a request time based at least in part on at least some of the fourth event data;
provide the request message comprising the request time to the communications interface via the first group of operator node(s); and
combine at least some of the third event data and at least some of the fourth event data.

2. The system of claim 1 , wherein the combiner node is configured to combine the at least some of the third event data and the at least some of the fourth event data by performing at least one of a join, temporal join, or union of the at least some of the third event data and the at least some of the fourth event data.

3. The system of claim 1 , wherein the combiner node is further configured to determine a timestamp associated with at least some of the fourth event data and to determine the request time based at least in part on the timestamp.

4. The system of claim 1 , wherein an operator node of the operator node(s) of the first group of operator node(s) is configured to:
receive the request message;
determine an altered request time based at least in part on the request time in the request message and a delay time introduced by the operator node;
determine an altered request message based at least in part on the request message and the altered request time; and
provide the altered request message to the communications interface in place of the request message.

5. The system of claim 4 , wherein at least one of the operator node(s) is configured to alter the request time based at least in part on a stored inverse function.

6. The system of claim 1 , wherein:
the source is configured to store the first event data associated with an interval index;
the request message comprises one or more times to be used to retrieve point and/or interval events from the store; and
an interval event comprises a start edge event and an end edge event.

7. The system of claim 1 , wherein the request message further comprises at least one of data of a number of events of the first event data to be retrieved from the source or data of a volume of the first event data to be retrieved from the source.

8. The system of claim 7 , wherein one or more of the operator node(s) of at least one of the first group of operator node(s) or the second group of operator node(s) are configured to modify the number of events to be retrieved from the source based at least in part on operations defined at the operator node(s).

9. The system of claim 1 , wherein at least one operator node of the first group of operator node(s) is configured to:
store a record of events received from at the at least one operator node;
receive the request message;
determine event data to be retrieved based at least in part on a difference between the record of events received at the at least one operator node and event data requested by the request message; and
generate and send an outgoing request message towards the communications interface, the outgoing request message requesting at least some of the determined event data to be retrieved.

10. The system of claim 1 , wherein at least one operator node of the first group of operator node(s) is configured to:
store a record of request messages received at the at least one operator node;
receive the request message; and
transmit the request message towards the communications interface based at least in part on the record indicating the request message is not a duplicate request message.

11. The system of claim 1 , wherein an operator node of the first group of operator node(s) is further configured to:
determine a time at which to provide the request message to the communications interface, the time determined such that the communications interface pulls the first event data responsive to the request message at such a time that the third event data and fourth event data reaches the combiner node simultaneously.

12. A method comprising:
receiving, at a combiner node, an event batch pushed to the combiner node from a communications interface over an event pipeline, the event batch being formed by the event pipeline from a live/replayed event stream received from the communications interface and pushed downstream the event pipeline;
responsive to receiving the event batch, determining, by the combiner node and based on at least some data of the event batch, a request time;
providing, by the combiner node, a request message comprising the request time to the communications interface upstream the event pipeline;
receiving, at the combiner node, data of historical events pulled from at least one event source by the communications interface in response to the request message, the combiner node providing the request message at an injection time determined by the combiner node such that the combiner node simultaneously receives the data of historical events and data from the live/replayed events corresponding with the request message; and
combining, at the combiner node, at least some of the data of the live/replayed events and at least some of the data of the historical events.

13. The method of claim 12 , wherein providing the request message from the combiner node to the communications interface upstream the event pipeline includes providing the request message to one or more operator nodes of the event pipeline, and the method of claim 12 further including:
determining an altered request time based at least in part on the request time in the request message and a delay time introduced by the one or more operator nodes, the determining including analyzing of a configuration of the one or more operator nodes; and
generating and sending an outgoing request message to the communications interface in place of the request message, the outgoing request message comprising the altered request time.

14. The method of claim 12 , further comprising:
providing the request message from the combiner node to the communications interface via an operator node of the event pipeline; and
at the operator node storing a record of event data received at the operator node and, on receipt of the request message:
determining event data to be retrieved based at least in part on a difference between a record of received or retrieved event data pushed to the operator node and event data requested by the incoming request message, and
generating and sending an outgoing request message towards the communicatio

Assignees

• Microsoft Technology Licensing Llc

Inventors

• Bussard Laurent
• Santos Ivo José Garcia Dos
• Nano Olivier
• Tarnavski Tihomir
• Goldstein Jonathan
• Chandramouli Badrish
• Novik Lev

Classifications

• G06F9/54 Primary

  Interprogram communication · CPC title

• G06F9/542 Primary

  Event management; Broadcasting; Multicasting; Notifications · CPC title

• G06F17/30575

  Physics · mapped topic

• G06F17/30548

  Physics · mapped topic

• G06F17/30516

  Physics · mapped topic