Windowing in SQL In batch processing, a SQL query that summarizes a table may require the server to read through every row in the table before returning a result. How is a streaming pipeline able to summarize a series of events that can continue forever? The answer is windowing. Windows create subsets of your streams with a start time and end time. A pipeline can then count or sum the windowed data and output the final totals for each time period. These windows can be created based on either event time (when the event originally occurred) or processing time (when the pipeline reads the event). Event time is usually preferable. See the time in pipelines section of the pipelines documentation to understand why. To use event time, a watermark must be set on the input stream. See watermarks section of the streams documentation for more information. This page assumes that you are already familiar with how to use SQL’s "group by" clause to aggregate data in a batch setting. Table-Valued Functions (TVF) Window aggregations in Decodable SQL look much like standard SQL aggregations. They use functions like count and sum in the select portion of the statement and then define how records should be grouped using the group by clause. The difference is that the source stream in the from clause is replaced with a table supplied by a Table-Valued Function: insert into <output-stream> select window_start, window_end, count(*) as count from table(<table-valued-function>(...)) group by window_start, window_end A TVF transforms a stream into time-bounded buckets of events that can be aggregated. It must be wrapped in the table() function when used in a pipeline’s SQL. While Decodable doesn’t support outputting the results of a TVF directly, you can think about them as a version of the input data with the additional fields window_start and window_end. Decodable supports Table-Valued Functions that implement a number of windowing strategies including tumble, hop, and cumulate. Tumble Tumble creates adjoining windows of equal size. It counts each input event in exactly one window. Tumbling windows are useful for reporting where you want events to belong to a single window, like taking the aggregate of customer orders in the last hour. tumble(table <data>, descriptor(<timecol>), <size>) Argument Description data your input stream name timecol the name of your watermarked event time field size a duration specifying the width of the windows The duration that a window should be open is specified using interval syntax. For example interval '5' seconds. -- Tumble use case SQL insert into orders_hourly select window_start, window_end, count(*) as orders, sum(payment) as payment from table( tumble(table orders, descriptor(`timestamp`), interval '1' hours)) group by window_start, window_end Hop Hop creates windows of equal size which may overlap. For this reason, an input event may be present in more than one window. Hopping windows are also known as "sliding windows". A classic example of a hopping window implementation is to summarize recent web requests from various IP address ranges for 1-minute windows in the last 15 minutes. hop(table <data>, descriptor(<timecol>), <slide>, <size>) Argument Description data your input stream name timecol the name of your watermarked event time field slide a duration specifying the time between the start of sequential hopping windows size a duration specifying the width of the windows The duration that a window should be open is specified using interval syntax. For example interval '5' seconds. -- Hop use case SQL insert into recent_requests select window_start, window_end, ip_range, count(*) as requests, sum(bytes) as bytes from table( hop(table orders, descriptor(`timestamp`), interval '1' minutes, interval '15' minutes)) group by window_start, window_end, ip_range Cumulate Cumulate creates cumulative windows of increasing size from the same start time until a maximum window size is reached. At that point, the previous window is closed and a new one is opened. Input events may be counted in more than one window. A classic example of a cumulate window implementation is to provide an updated count of unique daily users every 5 minutes. cumulate(table <data>, descriptor(<timecol>), <step>, <size>) Argument Description data your input stream name timecol the name of your watermarked event time field step a duration specifying the difference in window size between the end of sequential cumulating windows size a duration specifying the maximum size of the cumulating windows (must be an integer multiple of step) The duration that a window should be open is specified using interval syntax. For example interval '5' seconds. -- Cumulate use case SQL insert into current_daily_active_users select window_start, window_end, count(distinct user_id) as unique_users from table( cumulate(table orders, descriptor(`timestamp`), interval '5' minutes, interval '1' day)) group by window_start, window_end Key Takeaways A window is a subset of your stream bounded by a start time and an end time. To perform aggregation functions like count and sum, you must specify a window to perform the aggregation over using a Table-Valued Function (TVF). To aggregate using event time, your input stream must have watermarks enabled. See administer/manage-schemas.adoc#_how_to_use_the_schema_view.adoc for more on watermarks. Decodable currently supports three windowing strategies: Tumble Hop Cumulate Web UI Example Here’s an example showing how we can use windowing to aggregate logs of http activity using the tumble TVF: -- SQL text box contents from screenshot provided here for your convenience insert into http_aggregated select window_start, window_end, original_path, sum(bytes_rcvd) as bytes_rcvd from table( tumble(table http_parsed, descriptor(`timestamp`), interval '5' seconds)) group by window_start, window_end, original_path CLI Example Here is the same example using the Decodable CLI: > decodable stream get cafb4d28 http_parsed id cafb4d28 description auto created output stream for pipeline [parse_http] schema 0 timestamp TIMESTAMP(3) 1 method STRING 2 original_path STRING 3 response_code INT 4 bytes_rcvd INT watermark `timestamp` as `timestamp` create time 2022-06-08T18:36:20Z update time 2022-06-08T18:36:53Z > decodable pipeline preview "select * from http_parsed" Submitting query... done! (took 4.71s) Waiting for results... {"bytes_recv":8515,"method":"DELETE","original_path":"/products/2","response_code":201,"timestamp":"2022-06-07 22:24:16"} {"bytes_recv":1399,"method":"GET","original_path":"/products/2","response_code":200,"timestamp":"2022-06-07 22:24:17"} {"bytes_recv":384,"method":"PATCH","original_path":"/products/2","response_code":500,"timestamp":"2022-06-07 22:24:17"} ... Records received: 10 Time to first record: 21.56s Total time: 21.56so > decodable pipeline preview "insert into http_aggregated select window_start, window_end, original_path, sum(bytes_rcvd) as bytes_rcvd from table( tumble(table http_parsed, descriptor(\`timestamp\`), interval '5' seconds)) group by window_start, window_end, original_path" Submitting query... done! (took 7.51s) Waiting for results... {"bytes_rcvd":43233,"original_path":"/","window_end":"2022-06-09 20:17:30","window_start":"2022-06-09 20:17:25"} {"bytes_rcvd":14714,"original_path":"/products/3","window_end":"2022-06-09 20:17:30","window_start":"2022-06-09 20:17:25"} {"bytes_rcvd":15108,"original_path":"/users/1","window_end":"2022-06-09 20:17:30","window_start":"2022-06-09 20:17:25"} {"bytes_rcvd":15938,"original_path":"/products/2","window_end":"2022-06-09 20:17:30","window_start":"2022-06-09 20:17:25"} {"bytes_rcvd":56841,"original_path":"/products/3","window_end":"2022-06-09 20:17:35","window_start":"2022-06-09 20:17:30"} {"bytes_rcvd":49969,"original_path":"/products/2","window_end":"2022-06-09 20:17:35","window_start":"2022-06-09 20:17:30"} {"bytes_rcvd":48778,"original_path":"/","window_end":"2022-06-09 20:17:35","window_start":"2022-06-09 20:17:30"} {"bytes_rcvd":54100,"original_path":"/users/1","window_end":"2022-06-09 20:17:35","window_start":"2022-06-09 20:17:30"} {"bytes_rcvd":71614,"original_path":"/products/2","window_end":"2022-06-09 20:17:40","window_start":"2022-06-09 20:17:35"} {"bytes_rcvd":53903,"original_path":"/","window_end":"2022-06-09 20:17:40","window_start":"2022-06-09 20:17:35"} Records received: 10 Time to first record: 28.74s Total time: 36.87s 📘 This page contains copied contents from the Apache Flink® documentation. See Credits page for details.