Table API and SQL queries have the same semantics regardless whether their input is a finite set of rows or an unbounded stream of table changes. In many cases, continuous queries on streaming input are able to compute accurate results that are identical to offline computed results. However, for some continuous queries you have to limit the size of the state they are maintaining in order to avoid to run out of storage while ingesting an unbounded stream of input. It depends on the characteristics of the input data and the query itself whether you need to limit the state size and whether and how it affects the accuracy of the computed results.
Flink’s Table API and SQL interface provide parameters to tune the accuracy and resource consumption of continuous queries. The parameters are specified via a
TableConfig object, which can be obtained from the
In the following we describe the parameters of the
TableConfig and how they affect the accuracy and resource consumption of a query.
Many queries aggregate or join records on one or more key attributes. When such a query is executed on a stream, the continuous query needs to collect records or maintain partial results per key. If the key domain of the input stream is evolving, i.e., the active key values are changing over time, the continuous query accumulates more and more state as more and more distinct keys are observed. However, often keys become inactive after some time and their corresponding state becomes stale and useless.
For example the following query computes the number of clicks per session.
sessionId attribute is used as a grouping key and the continuous query maintains a count for each
sessionId it observes. The
sessionId attribute is evolving over time and
sessionId values are only active until the session ends, i.e., for a limited period of time. However, the continuous query cannot know about this property of
sessionId and expects that every
sessionId value can occur at any point of time. It maintains a count for each observed
sessionId value. Consequently, the total state size of the query is continuously growing as more and more
sessionId values are observed.
The Idle State Retention Time parameters define for how long the state of a key is retained without being updated before it is removed. For the previous example query, the count of a
sessionId would be removed as soon as it has not been updated for the configured period of time.
By removing the state of a key, the continuous query completely forgets that it has seen this key before. If a record with a key, whose state has been removed before, is processed, the record will be treated as if it was the first record with the respective key. For the example above this means that the count of a
sessionId would start again at
There are two parameters to configure the idle state retention time:
The parameters are specified as follows:
Cleaning up state requires additional bookkeeping which becomes less expensive for larger differences of
maxTime. The difference between
maxTime must be at least 5 minutes.