Jark Wu created FLINK-21109:
-------------------------------
Summary: Introduce "retractAccumulators" interface for
AggregateFunction in Table/SQL API
Key: FLINK-21109
URL: https://issues.apache.org/jira/browse/FLINK-21109
Project: Flink
Issue Type: New Feature
Components: Table SQL / API
Reporter: Jark Wu
*Motivation*
The motivation is to improve the performance of hopping (sliding) windows.
Currently, we have paned (or called sliced) optimization for the hopping
windows in Table/SQL.
That each element will be accumulated into a single pane. And once a window is
fired,
we will merge multiple panes to get the window result.
For example, HOP(size=10s, slide=2s), a window [0, 10) consists of 5 panes [0,
2), [2, 4), [4, 6), [6, 8), [8, 10).
And each element will fall into a single pane, e.g. element with timestamp 3
will fall into pane [2, 4).
However, currently, the merging panes happen on JVM heap memory. For example,
when window [0, 10) is going to be fired,
we will retrieve the accumulators of the 5 panes and merge them into an
in-memory accumulator.
The performance is not good, because the number of panes may be very large when
the slide is small, e.g. 8640 panes when HOP(1day, 10s).
And the memory may OOM when the accumulator is very large, e.g. containing
count distinct.
Thus, I would like to introduce a "retractAccumulators()" method which is an
inverse method of "merge()".
With the "retractAccumulators()" method, we can reduce the time complexity from
O(N) to O(1).
For example, when window [10, 20) is going to be fired, then we only need to
retract accumulator of pane [8, 10)
and merge the accumulator of pane [18, 20) into the state of the last window
[8, 18).
This will be a great performance improvement to make the hopping window have
similar performance
with the tumbling window, no matter how small the slide is.
*Public Interface*
We will introduce a contract method "retractAccumulators" which is similar to
the "merge" method.
{code}
Retracts a group of accumulator instances from one accumulator instance. This
method is optional,
but implementing this method can greatly improve the performance of hopping
window aggregates.
Therefore, it is recommended to implement this method when using with hopping
windows.
param: accumulator the accumulator which will keep the retracted aggregate
results. It should
be noted that the accumulator may contain the previous
aggregated
results. Therefore users should not replace or clean this
instance in the
custom retractAccumulators method.
param: retractAccs an java.lang.Iterable pointed to a group of accumulators
that will be
retracted.
public void retractAccumulators(ACC accumulator, java.lang.Iterable<ACC>
retractAccs)
{code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
