Google Git
Sign in
apache / flink-web / d06d0fee8a2129511fb8a365257019f577cb813d / . / content / 2019 / 05 / 19 / state-ttl.html
blob: f194148b0559a662b78e81333a08634c8079801d [file] [log] [blame]
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="viewport" content="width=device-width, initial-scale=1">
<!-- The above 3 meta tags *must* come first in the head; any other head content must come *after* these tags -->
<title>Apache Flink: State TTL in Flink 1.8.0: How to Automatically Cleanup Application State in Apache Flink</title>
<link rel="shortcut icon" href="/favicon.ico" type="image/x-icon">
<link rel="icon" href="/favicon.ico" type="image/x-icon">
<!-- Bootstrap -->
<link rel="stylesheet" href="https://maxcdn.bootstrapcdn.com/bootstrap/3.4.1/css/bootstrap.min.css">
<link rel="stylesheet" href="/css/flink.css">
<link rel="stylesheet" href="/css/syntax.css">
<!-- Blog RSS feed -->
<link href="/blog/feed.xml" rel="alternate" type="application/rss+xml" title="Apache Flink Blog: RSS feed" />
<!-- jQuery (necessary for Bootstrap's JavaScript plugins) -->
<!-- We need to load Jquery in the header for custom google analytics event tracking-->
<script src="/js/jquery.min.js"></script>
<!-- HTML5 shim and Respond.js for IE8 support of HTML5 elements and media queries -->
<!-- WARNING: Respond.js doesn't work if you view the page via file:// -->
<!--[if lt IE 9]>
<script src="https://oss.maxcdn.com/html5shiv/3.7.2/html5shiv.min.js"></script>
<script src="https://oss.maxcdn.com/respond/1.4.2/respond.min.js"></script>
<![endif]-->
</head>
<body>
<!-- Main content. -->
<div class="container">
<div class="row">
<div id="sidebar" class="col-sm-3">
<!-- Top navbar. -->
<nav class="navbar navbar-default">
<!-- The logo. -->
<div class="navbar-header">
<button type="button" class="navbar-toggle collapsed" data-toggle="collapse" data-target="#bs-example-navbar-collapse-1">
<span class="icon-bar"></span>
<span class="icon-bar"></span>
<span class="icon-bar"></span>
</button>
<div class="navbar-logo">
<a href="/">
<img alt="Apache Flink" src="/img/flink-header-logo.svg" width="147px" height="73px">
</a>
</div>
</div><!-- /.navbar-header -->
<!-- The navigation links. -->
<div class="collapse navbar-collapse" id="bs-example-navbar-collapse-1">
<ul class="nav navbar-nav navbar-main">
<!-- First menu section explains visitors what Flink is -->
<!-- What is Stream Processing? -->
<!--
<li><a href="/streamprocessing1.html">What is Stream Processing?</a></li>
-->
<!-- What is Flink? -->
<li><a href="/flink-architecture.html">What is Apache Flink?</a></li>
<!-- What is Stateful Functions? -->
<li><a href="/stateful-functions.html">What is Stateful Functions?</a></li>
<!-- Use cases -->
<li><a href="/usecases.html">Use Cases</a></li>
<!-- Powered by -->
<li><a href="/poweredby.html">Powered By</a></li>
&nbsp;
<!-- Second menu section aims to support Flink users -->
<!-- Downloads -->
<li><a href="/downloads.html">Downloads</a></li>
<!-- Getting Started -->
<li class="dropdown">
<a class="dropdown-toggle" data-toggle="dropdown" href="#">Getting Started<span class="caret"></span></a>
<ul class="dropdown-menu">
<li><a href="https://ci.apache.org/projects/flink/flink-docs-release-1.11/getting-started/index.html" target="_blank">With Flink <small><span class="glyphicon glyphicon-new-window"></span></small></a></li>
<li><a href="https://ci.apache.org/projects/flink/flink-statefun-docs-release-2.1/getting-started/project-setup.html" target="_blank">With Flink Stateful Functions <small><span class="glyphicon glyphicon-new-window"></span></small></a></li>
<li><a href="/training.html">Training Course</a></li>
</ul>
</li>
<!-- Documentation -->
<li class="dropdown">
<a class="dropdown-toggle" data-toggle="dropdown" href="#">Documentation<span class="caret"></span></a>
<ul class="dropdown-menu">
<li><a href="https://ci.apache.org/projects/flink/flink-docs-release-1.11" target="_blank">Flink 1.11 (Latest stable release) <small><span class="glyphicon glyphicon-new-window"></span></small></a></li>
<li><a href="https://ci.apache.org/projects/flink/flink-docs-master" target="_blank">Flink Master (Latest Snapshot) <small><span class="glyphicon glyphicon-new-window"></span></small></a></li>
<li><a href="https://ci.apache.org/projects/flink/flink-statefun-docs-release-2.1" target="_blank">Flink Stateful Functions 2.1 (Latest stable release) <small><span class="glyphicon glyphicon-new-window"></span></small></a></li>
<li><a href="https://ci.apache.org/projects/flink/flink-statefun-docs-master" target="_blank">Flink Stateful Functions Master (Latest Snapshot) <small><span class="glyphicon glyphicon-new-window"></span></small></a></li>
</ul>
</li>
<!-- getting help -->
<li><a href="/gettinghelp.html">Getting Help</a></li>
<!-- Blog -->
<li><a href="/blog/"><b>Flink Blog</b></a></li>
<!-- Flink-packages -->
<li>
<a href="https://flink-packages.org" target="_blank">flink-packages.org <small><span class="glyphicon glyphicon-new-window"></span></small></a>
</li>
&nbsp;
<!-- Third menu section aim to support community and contributors -->
<!-- Community -->
<li><a href="/community.html">Community &amp; Project Info</a></li>
<!-- Roadmap -->
<li><a href="/roadmap.html">Roadmap</a></li>
<!-- Contribute -->
<li><a href="/contributing/how-to-contribute.html">How to Contribute</a></li>
<!-- GitHub -->
<li>
<a href="https://github.com/apache/flink" target="_blank">Flink on GitHub <small><span class="glyphicon glyphicon-new-window"></span></small></a>
</li>
&nbsp;
<!-- Language Switcher -->
<li>
<a href="/zh/2019/05/19/state-ttl.html">中文版</a>
</li>
</ul>
<ul class="nav navbar-nav navbar-bottom">
<hr />
<!-- Twitter -->
<li><a href="https://twitter.com/apacheflink" target="_blank">@ApacheFlink <small><span class="glyphicon glyphicon-new-window"></span></small></a></li>
<!-- Visualizer -->
<li class=" hidden-md hidden-sm"><a href="/visualizer/" target="_blank">Plan Visualizer <small><span class="glyphicon glyphicon-new-window"></span></small></a></li>
<hr />
<li><a href="https://apache.org" target="_blank">Apache Software Foundation <small><span class="glyphicon glyphicon-new-window"></span></small></a></li>
<li>
<style>
.smalllinks:link {
display: inline-block !important; background: none; padding-top: 0px; padding-bottom: 0px; padding-right: 0px; min-width: 75px;
}
</style>
<a class="smalllinks" href="https://www.apache.org/licenses/" target="_blank">License</a> <small><span class="glyphicon glyphicon-new-window"></span></small>
<a class="smalllinks" href="https://www.apache.org/security/" target="_blank">Security</a> <small><span class="glyphicon glyphicon-new-window"></span></small>
<a class="smalllinks" href="https://www.apache.org/foundation/sponsorship.html" target="_blank">Donate</a> <small><span class="glyphicon glyphicon-new-window"></span></small>
<a class="smalllinks" href="https://www.apache.org/foundation/thanks.html" target="_blank">Thanks</a> <small><span class="glyphicon glyphicon-new-window"></span></small>
</li>
</ul>
</div><!-- /.navbar-collapse -->
</nav>
</div>
<div class="col-sm-9">
<div class="row-fluid">
<div class="col-sm-12">
<div class="row">
<h1>State TTL in Flink 1.8.0: How to Automatically Cleanup Application State in Apache Flink</h1>
<p><i></i></p>
<article>
<p>19 May 2019 Fabian Hueske (<a href="https://twitter.com/fhueske">@fhueske</a>) &amp; Andrey Zagrebin </p>
<p>A common requirement for many stateful streaming applications is to automatically cleanup application state for effective management of your state size, or to control how long the application state can be accessed (e.g. due to legal regulations like the GDPR). The state time-to-live (TTL) feature was initiated in Flink 1.6.0 and enabled application state cleanup and efficient state size management in Apache Flink.</p>
<p>In this post, we motivate the State TTL feature and discuss its use cases. Moreover, we show how to use and configure it. We explain how Flink internally manages state with TTL and present some exciting additions to the feature in Flink 1.8.0. The blog post concludes with an outlook on future improvements and extensions.</p>
<h1 id="the-transient-nature-of-state">The Transient Nature of State</h1>
<p>There are two major reasons why state should be maintained only for a limited time. For example, let’s imagine a Flink application that ingests a stream of user login events and stores for each user the time of the last login to improve the experience of frequent visitors.</p>
<ul>
<li>
<p><strong>Controlling the size of state.</strong>
Being able to efficiently manage an ever-growing state size is a primary use case for state TTL. Oftentimes, data needs to be persisted temporarily while there is some user activity around it, e.g. web sessions. When the activity ends there is no longer interest in that data while it still occupies storage. Flink 1.8.0 introduces background cleanup of old state based on TTL that makes the eviction of no-longer-necessary data frictionless. Previously, the application developer had to take extra actions and explicitly remove useless state to free storage space. This manual clean up procedure was not only error prone but also less efficient than the new lazy method to remove state. Following our previous example of storing the time of the last login, this might not be necessary after some time because the user can be treated as “infrequent” later on.</p>
</li>
<li>
<p><strong>Complying with data protection and sensitive data requirements.</strong>
Recent developments around data privacy regulations, such as the General Data Protection Regulation (GDPR) introduced by the European Union, make compliance with such data requirements or treating sensitive data a top priority for many use cases and applications. An example of such use cases includes applications that require keeping data for a specific timeframe and preventing access to it thereafter. This is a common challenge for companies providing short-term services to their customers. The state TTL feature gives guarantees for how long an application can access state and hence can help to comply with data protection regulations.</p>
</li>
</ul>
<p>Both requirements can be addressed by a feature that periodically, yet continuously, removes the state for a key once it becomes unnecessary or unimportant and there is no requirement to keep it in storage any more.</p>
<h1 id="state-ttl-for-continuous-cleanup-of-application-state">State TTL for continuous cleanup of application state</h1>
<p>The 1.6.0 release of Apache Flink introduced the State TTL feature. It enabled developers of stream processing applications to configure the state of operators to expire and be cleaned up after a defined timeout (time-to-live). In Flink 1.8.0 the feature was extended, including continuous cleanup of old entries for both the RocksDB and the heap state backends (FSStateBackend and MemoryStateBackend), enabling a continuous cleanup process of old entries (according to the TTL setting).</p>
<p>In Flink’s DataStream API, application state is defined by a <a href="https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/stream/state/state.html#using-managed-keyed-state">state descriptor</a>. State TTL is configured by passing a <code>StateTtlConfiguration</code> object to a state descriptor. The following Java example shows how to create a state TTL configuration and provide it to the state descriptor that holds the last login time of a user as a <code>Long</code> value:</p>
<div class="highlight"><pre><code class="language-java"><span class="kn">import</span> <span class="nn">org.apache.flink.api.common.state.StateTtlConfig</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">org.apache.flink.api.common.time.Time</span><span class="o">;</span>
<span class="kn">import</span> <span class="nn">org.apache.flink.api.common.state.ValueStateDescriptor</span><span class="o">;</span>
<span class="n">StateTtlConfig</span> <span class="n">ttlConfig</span> <span class="o">=</span> <span class="n">StateTtlConfig</span>
<span class="o">.</span><span class="na">newBuilder</span><span class="o">(</span><span class="n">Time</span><span class="o">.</span><span class="na">days</span><span class="o">(</span><span class="mi">7</span><span class="o">))</span>
<span class="o">.</span><span class="na">setUpdateType</span><span class="o">(</span><span class="n">StateTtlConfig</span><span class="o">.</span><span class="na">UpdateType</span><span class="o">.</span><span class="na">OnCreateAndWrite</span><span class="o">)</span>
<span class="o">.</span><span class="na">setStateVisibility</span><span class="o">(</span><span class="n">StateTtlConfig</span><span class="o">.</span><span class="na">StateVisibility</span><span class="o">.</span><span class="na">NeverReturnExpired</span><span class="o">)</span>
<span class="o">.</span><span class="na">build</span><span class="o">();</span>
<span class="n">ValueStateDescriptor</span><span class="o">&lt;</span><span class="n">Long</span><span class="o">&gt;</span> <span class="n">lastUserLogin</span> <span class="o">=</span>
<span class="k">new</span> <span class="n">ValueStateDescriptor</span><span class="o">&lt;&gt;(</span><span class="s">&quot;lastUserLogin&quot;</span><span class="o">,</span> <span class="n">Long</span><span class="o">.</span><span class="na">class</span><span class="o">);</span>
<span class="n">lastUserLogin</span><span class="o">.</span><span class="na">enableTimeToLive</span><span class="o">(</span><span class="n">ttlConfig</span><span class="o">);</span></code></pre></div>
<p>Flink provides multiple options to configure the behavior of the state TTL functionality.</p>
<ul>
<li>
<p><strong>When is the Time-to-Live reset?</strong>
By default, the expiration time of a state entry is updated when the state is modified. Optionally, it can also be updated on read access at the cost of an additional write operation to update the timestamp.</p>
</li>
<li>
<p><strong>Can the expired state be accessed one last time?</strong>
State TTL employs a lazy strategy to clean up expired state. This can lead to the situation that an application attempts to read state which is expired but hasn’t been removed yet. You can configure whether such a read request returns the expired state or not. In either case, the expired state is immediately removed afterwards. While the option of returning expired state favors data availability, not returning expired state can be required for data protection regulations.</p>
</li>
<li>
<p><strong>Which time semantics are used for the Time-to-Live timers?</strong>
With Flink 1.8.0, users can only define a state TTL in terms of processing time. The support for event time is planned for future Apache Flink releases.</p>
</li>
</ul>
<p>You can read more about how to use state TTL in the <a href="https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/state/state.html#state-time-to-live-ttl">Apache Flink documentation</a>.</p>
<p>Internally, the State TTL feature is implemented by storing an additional timestamp of the last relevant state access, along with the actual state value. While this approach adds some storage overhead, it allows Flink to check for the expired state during state access, checkpointing, recovery, or dedicated storage cleanup procedures.</p>
<h1 id="taking-out-the-garbage">“Taking out the Garbage”</h1>
<p>When a state object is accessed in a read operation, Flink will check its timestamp and clear the state if it is expired (depending on the configured state visibility, the expired state is returned or not). Due to this lazy removal, expired state that is never accessed again will forever occupy storage space unless it is garbage collected.</p>
<p>So how can the expired state be removed without the application logic explicitly taking care of it? In general, there are different possible strategies to remove it in the background.</p>
<h2 id="keep-full-state-snapshots-clean">Keep full state snapshots clean</h2>
<p>Flink 1.6.0 already supported automatic eviction of the expired state when a full snapshot for a checkpoint or savepoint is taken. Note that state eviction is not applied for incremental checkpoints. State eviction on full snapshots must be explicitly enabled as shown in the following example:</p>
<div class="highlight"><pre><code class="language-java"><span class="n">StateTtlConfig</span> <span class="n">ttlConfig</span> <span class="o">=</span> <span class="n">StateTtlConfig</span>
<span class="o">.</span><span class="na">newBuilder</span><span class="o">(</span><span class="n">Time</span><span class="o">.</span><span class="na">days</span><span class="o">(</span><span class="mi">7</span><span class="o">))</span>
<span class="o">.</span><span class="na">cleanupFullSnapshot</span><span class="o">()</span>
<span class="o">.</span><span class="na">build</span><span class="o">();</span></code></pre></div>
<p>The local storage stays untouched but the size of the stored snapshot is reduced. The local state of an operator will only be cleaned up when the operator reloads its state from a snapshot, i.e. in case of recovery or when starting from a savepoint.</p>
<p>Due to these limitations, applications still need to actively remove state after it expired in Flink 1.6.0. To improve the user experience, Flink 1.8.0 introduces two more autonomous cleanup strategies, one for each of Flink’s two state backend types. We describe them below.</p>
<h2 id="incremental-cleanup-in-heap-state-backends">Incremental cleanup in Heap state backends</h2>
<p>This approach is specific to the Heap state backends (FSStateBackend and MemoryStateBackend). The idea is that the storage backend keeps a lazy global iterator over all state entries. Certain events, for instance state access, trigger an incremental cleanup. Every time an incremental cleanup is triggered, the iterator is advanced. The traversed state entries are checked and expired once are removed. The following code example shows how to enable incremental cleanup:</p>
<div class="highlight"><pre><code class="language-java"><span class="n">StateTtlConfig</span> <span class="n">ttlConfig</span> <span class="o">=</span> <span class="n">StateTtlConfig</span>
<span class="o">.</span><span class="na">newBuilder</span><span class="o">(</span><span class="n">Time</span><span class="o">.</span><span class="na">days</span><span class="o">(</span><span class="mi">7</span><span class="o">))</span>
<span class="c1">// check 10 keys for every state access</span>
<span class="o">.</span><span class="na">cleanupIncrementally</span><span class="o">(</span><span class="mi">10</span><span class="o">,</span> <span class="kc">false</span><span class="o">)</span>
<span class="o">.</span><span class="na">build</span><span class="o">();</span></code></pre></div>
<p>If enabled, every state access triggers a cleanup step. For every clean up step, a certain number of state entries are checked for expiration. There are two tuning parameters. The first defines the number of state entries to check for each cleanup step. The second parameter is a flag to trigger a cleanup step after each processed record, additionally to each state access.</p>
<p>There are two important caveats about this approach:
* The first one is that the time spent for the incremental cleanup increases the record processing latency.
* The second one should be practically negligible but still worth mentioning: if no state is accessed or no records are processed, expired state won’t be removed.</p>
<h2 id="rocksdb-background-compaction-to-filter-out-expired-state">RocksDB background compaction to filter out expired state</h2>
<p>If your application uses the RocksDB state backend, you can enable another cleanup strategy which is based on a Flink specific compaction filter. RocksDB periodically runs asynchronous compactions to merge state updates and reduce storage. The Flink compaction filter checks the expiration timestamp of state entries with TTL and discards all expired values.</p>
<p>The first step to activate this feature is to configure the RocksDB state backend by setting the following Flink configuration option: <code>state.backend.rocksdb.ttl.compaction.filter.enabled</code>. Once the RocksDB state backend is configured, the compaction cleanup strategy is enabled for a state as shown in the following code example:</p>
<div class="highlight"><pre><code class="language-java"><span class="n">StateTtlConfig</span> <span class="n">ttlConfig</span> <span class="o">=</span> <span class="n">StateTtlConfig</span>
<span class="o">.</span><span class="na">newBuilder</span><span class="o">(</span><span class="n">Time</span><span class="o">.</span><span class="na">days</span><span class="o">(</span><span class="mi">7</span><span class="o">))</span>
<span class="o">.</span><span class="na">cleanupInRocksdbCompactFilter</span><span class="o">()</span>
<span class="o">.</span><span class="na">build</span><span class="o">();</span></code></pre></div>
<p>Keep in mind that calling the Flink TTL filter slows down the RocksDB compaction.</p>
<h2 id="eager-state-cleanup-with-timers">Eager State Cleanup with Timers</h2>
<p>Another way to manually cleanup state is based on Flink timers. This is an idea that the community is currently evaluating for future releases. With this approach, a cleanup timer is registered for every state access. This approach is more predictable because state is eagerly removed as soon as it expires. However, it is more expensive because the timers consume storage along with the original state.</p>
<h1 id="future-work">Future work</h1>
<p>Apart from including the timer-based cleanup strategy, mentioned above, the Flink community has plans to further improve the state TTL feature. The possible improvements include adding support of TTL for event time scale (only processing time is supported at the moment) and enabling State TTL for queryable state.</p>
<p>We encourage you to join the conversation and share your thoughts and ideas in the <a href="https://issues.apache.org/jira/projects/FLINK/summary">Apache Flink JIRA board</a> or by subscribing to the Apache Flink dev mailing list. Feedback or suggestions are always appreciated and we look forward to hearing your thoughts on the Flink mailing lists.</p>
<h1 id="summary">Summary</h1>
<p>Time-based state access restrictions and controlling the size of application state are common challenges in the world of stateful stream processing. Flink’s 1.8.0 release significantly improves the State TTL feature by adding support for continuous background cleanup of expired state objects. The new clean up mechanisms relieve you from manually implementing state cleanup. They are also more efficient due to their lazy nature. State TTL gives you control over the size of your application state so that you can focus on the core logic of your applications.</p>
</article>
</div>
<div class="row">
<div id="disqus_thread"></div>
<script type="text/javascript">
/* * * CONFIGURATION VARIABLES: EDIT BEFORE PASTING INTO YOUR WEBPAGE * * */
var disqus_shortname = 'stratosphere-eu'; // required: replace example with your forum shortname
/* * * DON'T EDIT BELOW THIS LINE * * */
(function() {
var dsq = document.createElement('script'); dsq.type = 'text/javascript'; dsq.async = true;
dsq.src = '//' + disqus_shortname + '.disqus.com/embed.js';
(document.getElementsByTagName('head')[0] || document.getElementsByTagName('body')[0]).appendChild(dsq);
})();
</script>
</div>
</div>
</div>
</div>
</div>
<hr />
<div class="row">
<div class="footer text-center col-sm-12">
<p>Copyright © 2014-2019 <a href="http://apache.org">The Apache Software Foundation</a>. All Rights Reserved.</p>
<p>Apache Flink, Flink®, Apache®, the squirrel logo, and the Apache feather logo are either registered trademarks or trademarks of The Apache Software Foundation.</p>
<p><a href="/privacy-policy.html">Privacy Policy</a> &middot; <a href="/blog/feed.xml">RSS feed</a></p>
</div>
</div>
</div><!-- /.container -->
<!-- Include all compiled plugins (below), or include individual files as needed -->
<script src="https://maxcdn.bootstrapcdn.com/bootstrap/3.3.4/js/bootstrap.min.js"></script>
<script src="https://cdnjs.cloudflare.com/ajax/libs/jquery.matchHeight/0.7.0/jquery.matchHeight-min.js"></script>
<script src="/js/codetabs.js"></script>
<script src="/js/stickysidebar.js"></script>
<!-- Google Analytics -->
<script>
(function(i,s,o,g,r,a,m){i['GoogleAnalyticsObject']=r;i[r]=i[r]||function(){
(i[r].q=i[r].q||[]).push(arguments)},i[r].l=1*new Date();a=s.createElement(o),
m=s.getElementsByTagName(o)[0];a.async=1;a.src=g;m.parentNode.insertBefore(a,m)
})(window,document,'script','//www.google-analytics.com/analytics.js','ga');
ga('create', 'UA-52545728-1', 'auto');
ga('send', 'pageview');
</script>
</body>
</html>