Data flow windowing and triggering

• US 10,732,928 B1
• Filed: 06/08/2018
• Issued: 08/04/2020
• Est. Priority Date: 11/03/2014
• Status: Active Grant

• Current Assignee: Google LLC (Alphabet Inc.)
• Original Assignee: Google LLC (Alphabet Inc.)
• Inventors: Akidau, Tyler, Bradshaw, Robert, Chambers, Ben, Chambers, Craig, Lax, Reuven, Mills, Daniel, Perry, Frances
• Primary Examiner(s): Vostal, Ondrej C
• Application Number:
  US16/003,997
• Time in Patent Office: 788 Days
• Field of Search: 709224
• CPC Class Codes: G06F 7/00 Methods or arrangements for processing data by operating upon the order or content of the data handled logic circuits H03K19/00
• CPC Class Codes: H04L 47/00 Traffic control in data switching networks arrangements for detecting or preventing errors in the information received H04L1/00
• CPC Class Codes: H04L 47/28 in relation to timing considerations

Abstract

A method includes receiving data corresponding one of streaming data or batch data and a content of the received data for computation. The method also includes determining an event time of the data for slicing the data, determining a processing time to output results of the received data, and emitting at least a portion of the results of the received data based on the processing time and the event time.

Claims

1. A method comprising:

receiving, at data processing hardware, data corresponding to one of streaming data or batch data;

determining, by the data processing hardware, an event time of the data for slicing the data;

grouping, by the data processing hardware, a first subset of the data into a first window, the first window defining a first sub-event time;

aggregating, by the data processing hardware, a first aggregated result processed from the first subset of the data for the first window;

determining, by the data processing hardware, a first trigger time to;

emit the first aggregated result; and

maintain the first aggregated result in a persistent state;

when a next aggregated result of a second subset of the data associated with the first window emits after emitting the first aggregated result;

emitting a retraction of the first aggregated result from the persistent state; and

emitting a combined session result for the first window, the combined session result comprising a sum of the first aggregated result and the next aggregated result; and

when the received data corresponds to streaming data;

setting, by the data processing hardware, an input timestamp on an element of the streaming data;

when the input timestamp on the element occurs earlier than a watermark, determining, by the data processing hardware, the streaming data comprises late streaming data; and

one of;

dropping the late streaming data;

or

allowing the late streaming data by creating a duplicate window in an output for the late streaming data.

2. The method of claim 1, further comprising:

determining, by the data processing hardware, a content of the received data for computation; and

determining, by the data processing hardware, a processing time to output results of the received data using the data processing hardware.

3. The method of claim 1, wherein maintaining the first aggregated result in the persistent state comprises storing a copy of the first aggregated result in memory hardware.

4. The method of claim 1, further comprising grouping, by the data processing hardware, the second subsets of the data into windows based on the event time, the windows comprising one of:

fixed windows defined by a static time period, each fixed window applied across all of the data within the associated time period;

sliding windows defined by a time period and a slide period, each sliding window applied across all of the data within the associated time period and associated with a start time separated from a start time of an immediately successive window by the slide period;

session windows defined by a timeout gap, each session window applied across a subset of the data occurring within a span of time less than the associated timeout gap;

or

user-defined windows defined by a pair of functions.

5. The method of claim 1, further comprising:

assigning, by the data processing hardware, a mergable window for each element of the received data, each element including an associated input timestamp and each session window extending a predefined range of time beyond the input timestamp for the associated window;

merging, by the data processing hardware, two or more of the mergable windows belonging to a same key that overlap into a single merged window; and

setting, by the data processing hardware, an associated output timestamp for each element to a value greater than or equal to an earliest time in the associated merged window or the associated mergeable window.

6. The method of claim 5, wherein the single merged window includes a range of time greater than the predefined range of time.

7. The method of claim 1, further comprising:

grouping, by the data processing hardware, a first subset of the received data into a second window, the second window defining a second sub-event time of the first data subset;

aggregating, by the data processing hardware, a second result of the first data subset for the second window; and

determining, by the data processing hardware, a second trigger time to emit the second aggregated result of the first data subset, the second trigger time comprising at least one of;

every threshold number of seconds of a walltime;

after receiving a punctuation record that terminates the second window;

every threshold number of records;

after arbitrary user logic decides to trigger;

or

after an arbitrary combination of concrete triggers.

8. The method of claim 7, further comprising, when determining the first trigger time to emit the first aggregated result of the first data subset, discarding, by the data processing hardware, the first aggregated result from use when aggregating results of the second subsets of the received data.

9. The method of claim 7, further comprising:

receiving, at the data processing hardware, a late data point after grouping the first subset of the data into the second window, the late data point related to the second window; and

discarding, by the data processing hardware, the late data point.

10. The method of claim 7, further comprising:

receiving, at the data processing hardware, a late data point after grouping the first subset of the data into the second window, the late data point related to the second window; and

accumulating, by the data processing hardware, the late data point into the second window to refine the second aggregated result with the late data point.

11. The method of claim 7, further comprising:

receiving, at the data processing hardware, a late data point after grouping the first subset of the data into the second window, the late data point related to the second window;

aggregating, by the data processing hardware, a combined result of the first subset of the data and the late data point; and

emitting the combined result.

12. A system comprising:

data processing hardware; and

memory hardware in communication with the data processing hardware, the memory hardware storing instructions that when executed on the data processing hardware cause the data processing hardware to perform operations comprising;

receiving data corresponding to one of streaming data or batch data;

determining an event time of the data for slicing the data;

grouping a first subset of the data into a first window, the first window defining a first sub-event time;

aggregating a first aggregated result processed from the first subset of the data for the first window; and

determining a first trigger time to;

emit the first aggregated result; and

maintain the first aggregated result in a persistent state; and

when a next aggregated result of a second subset of the data associated with the first window emits after emitting the first aggregated result;

emitting a retraction of the first aggregated result from the persistent state; and

emitting a combined session result for the first window, the combined session result comprising a sum of the first aggregated result and the next aggregated result; and

when the received data corresponds to streaming data;

setting an input timestamp on an element of the streaming data;

when the input timestamp on the element occurs earlier than a watermark, determining the streaming data comprises late streaming data; and

one of;

dropping the late streaming data;

or

allowing the late streaming data by creating a duplicate window in an output for the late streaming data.

13. The system of claim 12, wherein the operations further comprise:

determining a content of the received data for computation; and

determining a processing time to output results of the received data using the data processing hardware.

14. The system of claim 12, wherein maintaining the first aggregated result in the persistent state comprises storing a copy of the first aggregated result in the memory hardware.

15. The system of claim 12, wherein the operations further comprise grouping the second subsets of the data into windows based on the event time, the windows comprising one of:

fixed windows defined by a static time period, each fixed window applied across all of the data within the associated time period;

sliding windows defined by a time period and a slide period, each sliding window applied across all of the data within the associated time period and associated with a start time separated from a start time of an immediately successive window by the slide period;

session windows defined by a timeout gap, each session window applied across a subset of the data occurring within a span of time less than the associated timeout gap;

or

user-defined windows defined by a pair of functions.

16. The system of claim 12, wherein the operations further comprise:

assigning a mergable window for each element of the received data, each element including an associated input timestamp and each session window extending a predefined range of time beyond the input timestamp for the associated window;

merging two or more of the mergable windows belonging to a same key that overlap into a single merged window; and

setting an associated output timestamp for each element to a value greater than or equal to an earliest time in the associated merged window or the associated mergeable window.

17. The system of claim 16, wherein the single merged window includes a range of time greater than the predefined range of time.

18. The system of claim 12, wherein the operations further comprise:

grouping a first subset of the received data into a second window, the second window defining a second sub-event time of the first data subset;

aggregating a second result of the first data subset for the second window; and

determining a second trigger time to emit the second aggregated result of the first data subset, the second trigger time comprising at least one of;

every threshold number of seconds of a walltime;

after receiving a punctuation record that terminates the second window;

every threshold number of records;

after arbitrary user logic decides to trigger;

or

after an arbitrary combination of concrete triggers.

19. The system of claim 18, wherein the operations further comprise, when determining the first trigger time to emit the first aggregated result of the first data subset, discarding the first aggregated result from use when aggregating results of the second subsets of the received data.

20. The system of claim 18, wherein the operations further comprise:

receiving a late data point after grouping the first subset of the data into the second window, the late data point related to the second window; and

discarding the late data point.

21. The system of claim 18, wherein the operations further comprise:

receiving a late data point after grouping the first subset of the data into the second window, the late data point related to the second window; and

accumulating the late data point into the second window to refine the second aggregated result with the late data point.

22. The system of claim 18, wherein the operations further comprise:

receiving a late data point after grouping the first subset of the data into the second window, the late data point related to the second window;

aggregating a combined result of the first subset of the data and the late data point; and

emitting the combined result.