Data flow windowing and triggering

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.

What is claimed is: 
     
       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.