Variable duration windows on continuous data streams

• US 10,298,444 B2
• Filed: 03/15/2013
• Issued: 05/21/2019
• Est. Priority Date: 01/15/2013
• Status: Active Grant

• Current Assignee: Oracle International Corporation (Oracle Corporation)
• Original Assignee: Oracle International Corporation (Oracle Corporation)
• Inventors: Bishnoi, Sandeep, Srinivasan, Anand
• Primary Examiner(s): Wasel, Mohamed A
• Assistant Examiner(s): Kunwar, Binod J
• Application Number:
  US13/838,259
• Publication Number:
  US 20140201355A1
• Time in Patent Office: 2,258 Days
• Field of Search: 709224
• CPC Class Codes: H04L 41/0622 based on time
• CPC Class Codes: H04L 41/0686 Additional information in the notification, e.g. enhancement of specific meta-data
• CPC Class Codes: H04L 41/069 using logs of notifications; Post-processing of notifications

Abstract

Improved techniques are provided for processing streams of data. The duration of the time that an event is eligible for remaining in a time-based window can be variable for different events received via the same event stream. In certain embodiments, the duration of time that an input event spends in a time-based window is a function of one or more values of one or more attributes of the event. Since different events can have different one or more attribute values, the different events can spend different amounts of time in the time-based window for that event stream. The amount of time that an event spends in a time-based window can be controlled via the one or more attributes of the event.

Claims

1. A method comprising:

receiving, by a computing device, a query for an event stream, the event stream having an associated schema that each event received via the event stream conforms to, each event received via the event stream having one or more attributes associated with that event, the schema identifying how the one or more attributes for each event received via the event stream are organized, and the query specifying an expression indicating how an expiration duration is to be calculated for each event received via the event stream based on a first attribute of the one or more attributes associated with that event, the first attribute associated with each event identifiable based on the schema, the event stream generated by one or more event sources;

determining, by the computing device, a first expiration duration for a first event received via the event stream, the first expiration duration indicating a first amount of time that the first event is eligible to remain in a time-based window specified for the event stream, the first expiration duration determined based at least in part on the first attribute of the first event and the expression, the expression indicating how the first duration is to be calculated for the first event as specified in the query, the time-based window identifying a set of events of the event stream that are eligible to be queried by the query specified for the event stream;

determining, by the computing device, a second expiration duration for a second event received via the event stream, the second expiration duration indicating a second amount of time that the second event is eligible to remain in the time-based window, the second expiration duration determined based at least in part on the first attribute of the second event and the expression, the expression indicating how the second duration is to be calculated for the second event as specified in the query, wherein the second amount of time is different from the first amount of time;

removing, by the computing device, the first event from the time-based window after the first event has been in the time-based window for the first amount of time;

removing, by the computing device, the second event from the time-based window after the second event has been in the time-based window for the second amount of time;

repeatedly executing, by the computing device, the query against the set of events received via the event stream that are present within the time-based window when the query is executed; and

generating a set of one or more output events based on executing the query against the set of events.

2. The method of claim 1 wherein:

the first event has an associated first time, the first time corresponding to a time of receipt of the first event by the computing device that received the first event via the event stream;

the second event has an associated second time, the second time corresponding to a time of receipt of the second event by the computing device that received the second event via the event stream; and

the first time is same as the second time.

3. The method of claim 1 wherein:

the first event has an associated first time;

the second event has an associated second time; and

the first time is different from the second time.

4. The method of claim 1 wherein:

the expression specified in the query identifies the first attribute or the second attribute from the one or more attributes;

determining the first expiration duration for the first event is based upon a first value of the first attribute of the first event; and

determining the second expiration duration for the second event is based upon a second value of the first attribute of the second event, the second value being different from the first value.

5. The method of claim 1 further comprising:

determining, based upon a value of a second attribute of the first event, a first partition for the first event; and

determining, based upon a value of the second attribute of the second event, that the second event belongs to the first partition;

wherein the time-based window is for the first partition.

6. The method of claim 2, further comprising:

determining, by the computing device, a first expiration time for the first event based upon the first time associated with the first event and the first expiration duration that the first event is eligible to remain in the time-based window, the first expiration duration determined based upon a first value of the first attribute of the first event; and

determining, by the computing device, a second expiration time for the second event based upon the second time associated with the second event and the second expiration duration that the second event is eligible to remain in the time-based window, the second expiration duration determined based upon a second value of the first attribute of the second event, the second value being different from the first value; and

wherein;

removing the first event from the time-based window comprises removing the first event from the time-based window at or after the occurrence of the first expiration time; and

removing the second event from the time-based window comprises removing the second event from the time-based window at or after the occurrence of the second expiration time, the second expiration time being different from the first expiration time.

18. The method of claim 1, wherein the second event is received with or after the first event, and wherein the method further comprises:

when the first amount of time is larger than the second amount of time and a difference between the first amount of time and the second amount of time exceeds a time difference between the receipt of the first event and the second event, removing the second event from the time-based window before removing the first event from the time-based window.

19. The method of claim 1, wherein:

the second event is received with or after the first event;

the first amount of time that the first event is eligible to remain in the time-based window is larger than the second amount of time that the second event is eligible to remain in the time-based window; and

the set of events of the event stream that are within the time-based window when the query is executed includes the first event but not the second event.

7. A computing device comprising:

a memory; and

a set of processing units, wherein one or more processing units from the set of processing units are adapted to;

receive a query for an event stream, the event stream having an associated schema that each event received via the event stream conforms to, each event received via the event stream having one or more attributes associated with that event, the schema identifying how the one or more attributes for each event received via the event stream are organized, and the query specifying an expression indicating how an expiration duration is to be calculated for each event received via the event stream based on a first attribute of the one or more attributes associated with that event, the first attribute associated with each event identifiable based on the schema, the event stream generated by one or more event sources;

determine a first expiration duration for a first event received via the event stream, the first expiration duration indicating a first amount of time that the first event is eligible to remain in a time-based window specified for the event stream, the first expiration duration determined based at least in part on the first attribute of the first event and the expression, the expression indicating how the first duration is to be calculated for the first event as specified in the query, the time-based window identifying a set of events of the event stream that are eligible to be queried by the query specified for the event stream;

determine a second expiration duration for a second event received via the event stream, the second expiration duration indicating a second amount of time that the second event is eligible to remain in the time-based window, the second expiration duration determined based at least in part on the first attribute of the second event and the expression, the expression indicating how the second duration is to be calculated for the second event as specified in the query, wherein the second amount of time is different from the first amount of time;

remove the first event from the time-based window after the first event has been in the time-based window for the first amount of time;

remove the second event from the time-based window after the second event has been in the time-based window for the second amount of time;

repeatedly execute the query against the set of events received via the event stream that are present within the time-based window when the query is executed; and

generate a set of one or more output events based on executing the query against the set of events.

8. The computing device of claim 7 wherein:

the first event has an associated first time, the first time corresponding to a time of receipt of the first event by a processing unit that received the first event via the event stream;

the second event has an associated second time, the second time corresponding to a time of receipt of the second event by a processing unit that received the first event via the event stream; and

the first time is same as the second time.

9. The computing device of claim 7 wherein:

the first event has an associated first time;

the second event has an associated second time; and

the first time is different from the second time.

10. The computing device of claim 7 wherein the one or more processing units from the set of processing units are further adapted to:

identify a first attribute or the second attribute from one or more attributes from the expression specified in the query;

determine the first expiration duration for the first event based upon a first value of the first attribute of the first event; and

determine the second expiration duration for the second event based upon a second value of the first attribute of the second event, the second value being different from the first value.

11. The computing device of claim 7 wherein the one or more processing units from the set of processing units are further adapted to:

determine, based upon a value of a second attribute of the first event, a first partition for the first event; and

determine, based upon a value of the second attribute of the second event, that the second event belongs to the first partition;

wherein the time-based window is for the first partition.

12. The computing device of claim 8 wherein the one or more processing units from the set of processing units are further adapted to:

determine a first expiration time for the first event based upon the first time associated with the first event and the first expiration duration that the first event is eligible to remain in the time-based window, the first expiration duration determined based upon a first value of the first attribute of the first event; and

determine a second expiration time for the second event based upon the second time associated with the second event and the second expiration duration that the second event is eligible to remain in the time-based window, the second expiration duration determined based upon a second value of the first attribute of the second event, the second value being different from the first value; and

wherein;

when removing the first event from the time-based window, the one or more processing units from the set of processing units are further adapted to remove the first event from the time-based window at or after the occurrence of the first expiration time; and

when removing the second event from the time-based window, the one or more processing units from the set of processing units are further adapted to remove the second event from the time-based window at or after the occurrence of the second expiration time, the second expiration time being different from the first expiration time.

13. A computer-readable memory storing a plurality of instructions executable by one or more processing units, the plurality of instructions comprising:

instructions that cause at least one processing unit from the one or more processing units to receive a query for an event stream, the event stream having an associated schema that each event received via the event stream conforms to, each event received via the event stream having one or more attributes associated with that event, the schema identifying how the one or more attributes for each event received via the event stream are organized, and the query specifying an expression indicating how an expiration duration is to be calculated for each event received via the event stream based on a first attribute of the one or more attributes associated with that event, the first attribute associated with each event identifiable based on the schema, the event stream generated by one or more event sources;

instructions that cause at least one processing unit from the one or more processing units to determine a first expiration duration for a first event received via the event stream, the first expiration duration indicating a first amount of time that the first event is eligible to remain in a time-based window specified for the event stream, the first expiration duration determined based at least in part on the first attribute of the first event and the expression, the expression indicating how the first duration is to be calculated for the first event as specified in the query, the time-based window identifying a set of events of the event stream that are eligible to be queried by the query specified for the event stream;

instructions that cause at least one processing unit from the one or more processing units to determine a second expiration duration for a second event received via the event stream, the second expiration duration indicating a second amount of time that the second event is eligible to remain in the time-based window, the second duration determined based at least in part on the first attribute of the second event and the expression, the expression indicating how the second duration is to be calculated for the second event as specified in the query, wherein the second amount of time is different from the first amount of time;

instructions that cause at least one processing unit from the one or more processing units to remove the first event from the time-based window after the first event has been in the time-based window for the first amount of time;

instructions that cause at least one processing unit from the one or more processing units to remove the second event from the time-based window after the second event has been in the time-based window for the second amount of time;

instructions that cause at least one processing unit from the one or more processing units to repeatedly execute the query against the set of events received via the event stream that are present within the time-based window when the query is executed; and

instructions that cause at least one processing unit from the one or more processing units to generate a set of one or more output events based on executing the query against the set of events.

14. The computer-readable memory of claim 13 wherein:

the first event has an associated first time, the first time corresponding to a time of receipt of the first event by the processing unit that received the first event via the event stream;

the second event has an associated second time, the second time corresponding to a time of receipt of the second event by the processing unit that received the first event via the event stream; and

the first time is same as the second time.

15. The computer-readable memory of claim 13 wherein the plurality of instructions further comprises:

instructions that cause at least one processing unit from the one or more processing units to identify a first attribute from one or more attributes from the expression specified in the query;

instructions that cause at least one processing unit from the one or more processing units to determine the first duration for the first event based upon a first value of the first attribute of the first event; and

instructions that cause at least one processing unit from the one or more processing units to determine the second duration for the second event based upon a second value of the first attribute of the second event, the second value being different from the first value.

16. The computer-readable memory of claim 13 wherein the plurality of instructions further comprises:

instructions that cause at least one processing unit from the one or more processing units to determine, based upon a value of a second attribute of the first event, a first partition for the first event; and

instructions that cause at least one processing unit from the one or more processing units to determine, based upon a value of the second attribute of the second event, that the second event belongs to the first partition;

wherein the time-based window is for the first partition.

17. The computer-readable memory of claim 14 wherein the plurality of instructions further comprises:

instructions that cause at least one processing unit from the one or more processing units to determine a first expiration time for the first event based upon the first time associated with the first event and the first expiration duration of time that the first event is eligible to remain in the time-based window, the first expiration duration determined based upon a first value of the first attribute of the first event;

instructions that cause at least one processing unit from the one or more processing units to determine a second expiration time for the second event based upon the second time associated with the second event and the second expiration duration of time that the second event is eligible to remain in the time-based window, the second expiration duration determined based upon a second value of the first attribute of the second event, the second value being different from the first value;

instructions that cause at least one processing unit from the one or more processing units, when removing the first event from the time-based window, to remove the first event from the time-based window at or after the occurrence of the first expiration time; and

instructions that cause at least one processing unit from the one or more processing units, when removing the second event from the time-based window, to remove the second event from the time-based window at or after the occurrence of the second expiration time, the second expiration time being different from the first expiration time.