Status
Current state: Under Discussion
Discussion thread: here (<- link to https://mail-archives.apache.org/mod_mbox/flink-dev/)
JIRA: here (<- link to https://issues.apache.org/jira/browse/FLINK-XXXX)
Released: <Flink Version>
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Motivation
The main purpose of this FLIP is to improve the near-real-time (NRT) experience of Flink. We propose to support windowing table-valued function (TVF) syntax as an entry point of NRT use cases. We will explain why we have this decision and the benefits of introducing windowing TVF.
Usually, we can categorize data processing into real-time (in seconds), near real-time (in minutes), and batch (> hours) processing. Flink is a well-known stream processing system which is good at real-time scenarios. Meanwhile, the community has put a lot of effort into strengthening batch processing capabilities. However,
I have heard some users complained that it's expensive to use Flink for NRT use cases. As far as we know, NRT is a very common scenarios. We might have overlooked it before. That's why we want to improve it and make Flink to be a great engine for NRT scenarios.
Goal
What we need to do for that purpose? We investigated many Flink streaming jobs and we found the following pain points.
- Learning curve. Usually, users use windows for their minutes/seconds-granularity statistics. However, windows is not easy to use in Flink SQL currently. It only supports window aggregate, not support window join, window TopN, window deduplicate. It's hard to cascade different operations (e.g. join, agg), users have to learn how to keep time attribute and some streaming specific functions, e.g.
TUMBLE_ROWTIME
. - Performance. Flink is a native streaming engine, it can provide low latency with the cost of per-record state operation. But users don't need such a low latency in some cases. It would be great if the tolerated delay can be exchanged for a huge increase in throughput.
In the industry, users typically use batch engine and scheduler to build NRT pipelines. We also investigated lots of such jobs and found most of jobs are 15-minutes and cumulative aggregations (aggregations from 0 ~ current minute). For example, the number of cumulative UV at 10:00 represents the total number of UV from 00:00 to 10:00. Therefore, the daily report is a monotonically increasing line like the following picture. Snowflake also provides an example of cumulative window aggregates.
According to our research, most of such use cases are supported by a batch job and scheduled every 15 minutes. The upstream is a 15-minutes partitioned table with the detailed data in that 15 minutes. The batch job will load all the partitions so far today for aggregating using such filter where ds = '${current_date}' and (hh <'${hour}' or (hh ='${hour}' and mm <= '${minute}'))
. However, it has following pain points:
- Repeated calculation. The older partitions will be read and calculated more than once.
- Latency. The overhead of scheduler increases the delay. The resources is not guaranteed when cluster is busy. Usually, the latency of a 15-minutes job is more than 30 minutes.
- Unstable output of data. When the resource usage of cluster is in the peak, some partition scheduling tasks are lost.
- Data drift. Some data of 14:59 event time might be wroten into [15:00, 30:00) partition. Then the aggregation result for [00:00, 15:00) partition is not accurate. Users can read one more partition, but it increase latency again.
Based on the research above, the goal of improving NRT is to provide an easy to use API with good performance and reslove the problems users meet in their batch job + scheduler solution. This can extend the border of Flink SQL and attract users from traditional batch processing.
Windowing TVF is the solution we propose for this goal. We think with new API and improved performance, we can achieve the following advantages and address above pain points.
- Reduce latency. The stream processing and get rid of the scheduler which can save the scheduling time. Streaming system can continuously process, it can produce output in a more timely manner.
- Steady output of data. Streaming job is a long running task, the output of windows will never get lost.
- Solve data drift and out-of-order problem with the watermark mechanism.
- Reduce the learning curve and support richer operations for windowing. It's easier to propagate the regular window_start and window_end columns than the special time attribute column.
Public Interfaces
The idea of windowing table-valued function is from "One SQL To Rule Them All" which is published & presented at SIGMOD 2019. It leverages the new introduced Polymorphic Table Functions (abbreviated PTF) which are the part of the SQL 2016 standard. The "One SQL" paper illustrates two windowing TVFs: Tumbling and hopping windows. For Flink SQL, we would also like to add a new kind window for cumulative aggregations which is quite common in NRT.
This requires the parser support of Calcite. Fortunately, Calcite v1.25.0 has already supports tumbling and hopping windowing TVF. We need to extend it to support one more window: cumulative window.
Windowing table-valued functions
Tumbling Windows
The table-valued function TUMBLE
assigns a window for each row of a relation based on a time attribute column. The return value of TUMBLE
is a relation that includes all columns of data as well as additional two columns named window_start
and window_end
which indicate the assigned window. All assigned windows have the same length, and that’s why tumbling sometimes is named as “fixed windowing”.
TUMBLE
that takes three required parameters and one optional parameter:
Code Block | ||
---|---|---|
| ||
TUMBLE(data, DESCRIPTOR(timecol), size [, offset ]) |
data
is a table parameter that can be any relation with a time attribute column.timecol
is a column descriptor indicating which time attribute column of data should be mapped to tumbling windows.size
is a duration specifying the width of the tumbling windows.offset
(optional) specifies that the tumbling should begin from an instant other than the standard beginning of the epoch.
Here is an example invocation on the Bid table from the paper:
Code Block | ||
---|---|---|
| ||
> SELECT * FROM Bid;
--------------------------
| bidtime | price | item |
--------------------------
| 8:07 | $2 | A |
| 8:11 | $3 | B |
| 8:05 | $4 | C |
| 8:09 | $5 | D |
| 8:13 | $1 | E |
| 8:17 | $6 | F |
--------------------------
> SELECT * FROM TABLE(
TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES));
------------------------------------------------------
| bidtime | price | item | window_start | window_end |
------------------------------------------------------
| 8:07 | $2 | A | 8:00 | 8:10 |
| 8:11 | $3 | B | 8:10 | 8:20 |
| 8:05 | $4 | C | 8:00 | 8:10 |
| 8:09 | $5 | D | 8:00 | 8:10 |
| 8:13 | $1 | E | 8:10 | 8:20 |
| 8:17 | $6 | F | 8:10 | 8:20 |
------------------------------------------------------
> SELECT window_start, window_end, SUM(price)
FROM TABLE(
TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES))
GROUP BY window_start, window_end;
-------------------------------------
| window_start | window_end | price |
-------------------------------------
| 8:00 | 8:10 | $11 |
| 8:10 | 8:20 | $10 |
------------------------------------- |
Hopping Windows
The table-valued function HOP
assigns windows that cover rows within the interval of size and shifting every slide based on a timestamp column. The return value of HOP
is a relation that includes all columns of data as well as additional two columns named window_start
and window_end
which indicate the assigned window. Windows assigned could have overlapping so hopping sometime is named as “sliding windowing”.
HOP
takes four required parameters and one optional parameter.
Code Block | ||
---|---|---|
| ||
HOP(data, DESCRIPTOR(timecol), slide, size [, offset ]) |
data
is a table parameter that can be any relation with an time attribute column.timecol
is a column descriptor indicating which time attribute column of data should be mapped to tumbling windows.slide
is a duration specifying the duration between the start of sequential hopping windowssize
is a duration specifying the width of the hopping windows.offset
(optional) specifies that the hopping should begin from an instant other than the standard beginning of the epoch.
Here is an example invocation on the Bid table from the paper:
Code Block | ||
---|---|---|
| ||
> SELECT * FROM TABLE(
HOP(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES));
------------------------------------------------------
| bidtime | price | item | window_start | window_end |
------------------------------------------------------
| 8:07 | $2 | A | 8:00 | 8:10 |
| 8:07 | $2 | A | 8:05 | 8:15 |
| 8:11 | $3 | B | 8:05 | 8:15 |
| 8:11 | $3 | B | 8:10 | 8:20 |
| 8:05 | $4 | C | 8:00 | 8:10 |
| 8:05 | $4 | C | 8:05 | 8:15 |
| 8:09 | $5 | D | 8:00 | 8:10 |
| 8:09 | $5 | D | 8:05 | 8:15 |
| 8:13 | $1 | E | 8:05 | 8:15 |
| 8:13 | $1 | E | 8:10 | 8:20 |
| 8:17 | $6 | F | 8:10 | 8:20 |
| 8:17 | $6 | F | 8:15 | 8:25 |
------------------------------------------------------
> SELECT window_start, window_end, SUM(price)
FROM TABLE(
HOP(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES))
GROUP BY window_start, window_end;
--------------------------------------
| window_start | window_end | price |
--------------------------------------
| 8:00 | 8:10 | $11 |
| 8:05 | 8:15 | $15 |
| 8:10 | 8:20 | $10 |
| 8:15 | 8:25 | $6 |
-------------------------------------- |
Cumulating Windows
The table-valued function CUMULATE
assigns windows that cover rows within an initial interval of step size, and expanding to one more step size (keep window start fixed) every step until to the max window size. The return value of CUMULATE
is a relation that includes all columns of data as well as additional two columns named window_start
and window_end
which indicate the assigned window. Windows assigned could have overlapping.
CUMULATE
takes four required parameters and one optional parameter.
Code Block | ||
---|---|---|
| ||
CUMULATE(data, DESCRIPTOR(timecol), step, size [, offset ]) |
data
is a table parameter that can be any relation with an time attribute column.timecol
is a column descriptor indicating which time attribute column of data should be mapped to tumbling windows.step
is a duration specifying the increased window size between the end of sequential cumulating windows.size
is a duration specifying the max width of the cumulating windows.size
must be an integral multiple ofstep
.offset
(optional) specifies that the hopping should begin from an instant other than the standard beginning of the epoch.
So a cumulating window for 1 hour step and 1 day max size will produce wndows: [00:00, 01:00), [00:00, 02:00), [00:00, 03:00), ..., [00:00, 24:00) for every day.
Here is an example invocation on the Bid table:
Code Block | ||
---|---|---|
| ||
> SELECT * FROM TABLE(
CUMULATE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES));
------------------------------------------------------
| bidtime | price | item | window_start | window_end |
------------------------------------------------------
| 8:07 | $2 | A | 8:00 | 8:08 |
| 8:07 | $2 | A | 8:00 | 8:10 |
| 8:11 | $3 | B | 8:10 | 8:12 |
| 8:11 | $3 | B | 8:10 | 8:14 |
| 8:11 | $3 | B | 8:10 | 8:16 |
| 8:11 | $3 | B | 8:10 | 8:18 |
| 8:11 | $3 | B | 8:10 | 8:20 |
| 8:05 | $4 | C | 8:00 | 8:06 |
| 8:05 | $4 | C | 8:00 | 8:08 |
| 8:05 | $4 | C | 8:00 | 8:10 |
| 8:09 | $5 | D | 8:00 | 8:10 |
| 8:13 | $1 | E | 8:10 | 8:14 |
| 8:13 | $1 | E | 8:10 | 8:16 |
| 8:13 | $1 | E | 8:10 | 8:18 |
| 8:13 | $1 | E | 8:10 | 8:20 |
| 8:17 | $6 | F | 8:10 | 8:18 |
| 8:17 | $6 | F | 8:10 | 8:20 |
------------------------------------------------------
> SELECT window_start, window_end, SUM(price)
FROM TABLE(
CUMULATE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES))
GROUP BY window_start, window_end;
--------------------------------------
| window_start | window_end | price |
--------------------------------------
| 8:00 | 8:06 | $4 |
| 8:00 | 8:08 | $6 |
| 8:00 | 8:10 | $11 |
| 8:10 | 8:12 | $3 |
| 8:10 | 8:14 | $4 |
| 8:10 | 8:16 | $4 |
| 8:10 | 8:18 | $10 |
| 8:10 | 8:20 | $10 |
-------------------------------------- |
Operations on windows
We also would like to support various operations based on the windowing TVFs. This can make the functionality more complete.
Window Aggregate
The window aggregate requires the group by clause contains start and end columns of the windowing TVF.
Code Block | ||
---|---|---|
| ||
SELECT ...
FROM T -- relation applied windowing TVF
GROUP BY window_start, window_end, ... |
In the future, we can also simplify the group by clause to only include the start or end column if the windowing TVF is TUMBLE
or HOP
.
Window Join
The window join requires the join on condition contains window starts equality of input tables and window ends equality of input tables. The semantic of window join is the same to the DataStream window join.
SELECT ... FROM L [LEFT|RIGHT] JOIN R -- L and R are relations applied windowing TVF ON L.window_start = R.window_start AND L.window_end = R.window_end AND ...
In the future, we can also simplify the join on clause to only include the window start equality if the windowing TVF is TUMBLE
or HOP
. Currently, the windowing TVFs must be the same of left and right inputs. This can be extended in the future, for example, tumbling windows join sliding windows with the same window size.
Window TopN
The window TopN requires the PARTITION BY clause contains start and end columns of the windowing TVF.
SELECT ... FROM ( SELECT *, ROW_NUMBER() OVER (PARTITION BY window_start, window_end, ... ORDER BY col1 [asc|desc][, col2 [asc|desc]...]) AS rownum FROM T) -- relation applied windowing TVF WHERE rownum <= N [AND conditions]
In the future, we can also simplify the PARTITION BY clause to only include the start or end column if the windowing TVF is TUMBLE
or HOP
.
This is a general FLIP to describe the background and syntax of the window table functions we introduce, such as the window join, window aggregation and so on.
There would be separate sub FLIP for each kind of the logical operations on the window table functions.
The Syntax
...
language | sql |
---|
...