Flink系列之：Flink 1.8.0 中的状态 TTL：如何在 Apache Flink 中自动清理应用程序状态

许多有状态流应用程序的常见要求是自动清理应用程序状态，以有效管理状态大小，或控制应用程序状态可以访问的时间（例如，由于 GDPR 等法律法规）。状态生存时间 (TTL) 功能在 Flink 1.6.0 中启动，并在 Apache Flink 中启用应用程序状态清理和高效的状态大小管理。

在这篇文章中，我们将激发 State TTL 功能并讨论其用例。此外，我们还展示了如何使用和配置它。我们解释了 Flink 如何在内部使用 TTL 管理状态，并介绍了 Flink 1.8.0 中该功能的一些令人兴奋的新增功能。这篇博文最后展望了未来的改进和扩展。

一、状态的瞬态性质

状态只能维持有限的时间有两个主要原因。例如，让我们想象一个 Flink 应用程序，它摄取用户登录事件流并为每个用户存储上次登录的时间，以改善频繁访问者的体验。

• 控制状态的大小。能够有效管理不断增长的状态大小是状态 TTL 的主要用例。通常，当数据周围有一些用户活动时，数据需要暂时保留，例如网络会话。当活动结束时，人们不再对该数据感兴趣，但它仍然占用存储空间。 Flink 1.8.0 引入了基于 TTL 的旧状态后台清理功能，可以轻松驱逐不再需要的数据。以前，应用程序开发人员必须采取额外的操作并显式删除无用的状态以释放存储空间。这种手动清理过程不仅容易出错，而且比新的惰性删除状态方法效率低。按照我们之前存储上次登录时间的示例，一段时间后可能不需要这样做，因为稍后可以将用户视为“不频繁”。
• 遵守数据保护和敏感数据要求。围绕数据隐私法规的最新发展，例如欧盟推出的通用数据保护法规 (GDPR)，使得遵守此类数据要求或处理敏感数据成为许多用例和应用程序的首要任务。此类用例的一个示例包括需要在特定时间范围内保留数据并在此后阻止对其进行访问的应用程序。对于向客户提供短期服务的公司来说，这是一个常见的挑战。状态 TTL 功能保证了应用程序可以访问状态的时间长度，因此有助于遵守数据保护法规。

这两个要求都可以通过一个功能来解决，该功能一旦密钥变得不必要或不重要，就会定期但连续地删除密钥的状态，并且不再需要将其保留在存储中。

二、用于持续清理应用程序状态的状态 TTL

Apache Flink 1.6.0 版本引入了 State TTL 功能。它使流处理应用程序的开发人员能够将运算符的状态配置为在定义的超时（生存时间）后过期并清除。在 Flink 1.8.0 中，该功能得到了扩展，包括持续清理 RocksDB 和堆状态后端（FSStateBackend 和 MemoryStateBackend）的旧条目，从而实现旧条目的持续清理过程（根据 TTL 设置）。

在 Flink 的 DataStream API 中，应用程序状态由状态描述符定义。状态 TTL 是通过将 StateTtlConfiguration 对象传递给状态描述符来配置的。以下 Java 示例演示如何创建状态 TTL 配置并将其提供给状态描述符，该状态描述符将用户的上次登录时间保存为 Long 值：

import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.state.ValueStateDescriptor;

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.days(7))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .build();
    
ValueStateDescriptor<Long> lastUserLogin = 
    new ValueStateDescriptor<>("lastUserLogin", Long.class);

lastUserLogin.enableTimeToLive(ttlConfig);

这段代码使用Apache Flink提供的StateTtlConfig来设置状态的TTL（Time-To-Live）配置。

• 首先，导入必要的包org.apache.flink.api.common.state.StateTtlConfig、org.apache.flink.api.common.time.Time和org.apache.flink.api.common.state.ValueStateDescriptor。
• 然后，创建StateTtlConfig对象ttlConfig，并使用StateTtlConfig.newBuilder(Time.days(7))来指定TTL的时间长度为7天。这意味着状态数据的最大生存时间为7天。
• 接下来，调用ttlConfig的setUpdateType方法，将UpdateType设置为StateTtlConfig.UpdateType.OnCreateAndWrite。这表示在创建和写入状态时更新TTL。
• 然后，调用ttlConfig的setStateVisibility方法，将StateVisibility设置为StateTtlConfig.StateVisibility.NeverReturnExpired。这表示状态在过期后永远不会返回，也就是被清理后不会再被读取。
• 最后，使用ValueStateDescriptor创建一个名为"lastUserLogin"的状态描述符lastUserLogin，并调用lastUserLogin的enableTimeToLive方法，将ttlConfig传递给它。这将启用状态的TTL配置。
• 通过配置TTL，可以控制状态的生存时间，以及何时更新和清理状态。这有助于管理状态数据的存储和性能。在这个例子中，状态"lastUserLogin"的过期时间为7天，并且在创建和写入状态时更新TTL。

Flink 提供了多个选项来配置状态 TTL 功能的行为。

• 生存时间何时重置？默认情况下，状态条目的过期时间会在状态修改时更新。或者，也可以在读取访问时更新它，但需要执行额外的写入操作来更新时间戳。
• 过期状态可以最后一次访问吗？状态 TTL 采用惰性策略来清理过期状态。这可能会导致应用程序尝试读取已过期但尚未删除的状态。您可以配置此类读取请求是否返回过期状态。无论哪种情况，过期状态都会立即被删除。虽然返回过期状态的选项有利于数据可用性，但数据保护法规可能需要不返回过期状态。
• 生存时间计时器使用哪些时间语义？在 Flink 1.8.0 中，用户只能根据处理时间定义状态 TTL。计划在未来的 Apache Flink 版本中支持事件时间。

在内部，状态 TTL 功能是通过存储最后一个相关状态访问的附加时间戳以及实际状态值来实现的。虽然这种方法增加了一些存储开销，但它允许 Flink 在状态访问、检查点、恢复或专用存储清理过程期间检查过期状态。

三、倒垃圾

当读操作访问状态对象时，Flink 会检查其时间戳，如果过期则清除状态（根据配置的状态可见性，是否返回过期状态）。由于这种惰性删除，不再被访问的过期状态将永远占用存储空间，除非它被垃圾收集。

那么，在应用程序逻辑不明确处理的情况下，如何删除过期状态呢？一般来说，有不同的可能策略可以在后台将其删除。

四、保持完整状态快照干净

Flink 1.6.0 已经支持在拍摄检查点或保存点的完整快照时自动驱逐过期状态。请注意，状态驱逐不适用于增量检查点。必须显式启用完整快照的状态驱逐，如下例所示：

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.days(7))
    .cleanupFullSnapshot()
    .build();

本地存储保持不变，但存储快照的大小减小。仅当操作员从快照重新加载其状态时，即在恢复或从保存点启动时，操作员的本地状态才会被清除。

由于这些限制，在 Flink 1.6.0 中应用程序仍然需要在状态过期后主动删除状态。为了改善用户体验，Flink 1.8.0 引入了两种更多的自主清理策略，针对 Flink 的两种状态后端类型各一种。我们在下面描述它们。

五、堆状态后端的增量清理

此方法特定于堆状态后端（FSStateBackend 和 MemoryStateBackend）。这个想法是存储后端在所有状态条目上保留一个惰性全局迭代器。某些事件（例如状态访问）会触发增量清理。每次触发增量清理时，迭代器都会前进。遍历的状态条目会被检查，一旦被删除就会过期。以下代码示例展示了如何启用增量清理：

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.days(7))
    // check 10 keys for every state access
    .cleanupIncrementally(10, false)
    .build();

如果启用，每个状态访问都会触发清理步骤。对于每个清理步骤，都会检查一定数量的状态条目是否过期。有两个调整参数。第一个定义了每个清理步骤要检查的状态条目数。第二个参数是一个标志，用于在每个处理的记录之后以及每个状态访问之后触发清理步骤。

这种方法有两个重要的注意事项：

• 第一个是增量清理所花费的时间增加了记录处理延迟。
• 第二个几乎可以忽略不计，但仍然值得一提：如果没有访问任何状态或没有处理任何记录，则不会删除过期状态。

六、RocksDB 后台压缩以过滤掉过期状态

如果您的应用程序使用 RocksDB 状态后端，您可以启用另一种基于 Flink 特定压缩过滤器的清理策略。 RocksDB 定期运行异步压缩来合并状态更新并减少存储。 Flink 压缩过滤器使用 TTL 检查状态条目的过期时间戳，并丢弃所有过期值。

激活此功能的第一步是通过设置以下 Flink 配置选项来配置 RocksDB 状态后端：state.backend.rocksdb.ttl.compaction.filter.enabled。配置 RocksDB 状态后端后，将为状态启用压缩清理策略，如以下代码示例所示：

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.days(7))
    .cleanupInRocksdbCompactFilter()
    .build();

请记住，调用 Flink TTL 过滤器会减慢 RocksDB 压缩速度。

七、使用计时器进行急切的状态清理

另一种手动清理状态的方法是基于 Flink 计时器。社区目前正在评估这个想法以用于未来的版本。通过这种方法，为每个状态访问注册一个清理计时器。这种方法更具可预测性，因为状态一旦过期就会被立即删除。然而，它更昂贵，因为计时器消耗存储以及原始状态。

八、未来的工作

除了上面提到的基于定时器的清理策略之外，Flink 社区还计划进一步改进状态 TTL 功能。可能的改进包括添加对事件时间尺度的 TTL 支持（目前仅支持处理时间）以及为可查询状态启用状态 TTL。

九、总结

基于时间的状态访问限制和控制应用程序状态的大小是有状态流处理领域的常见挑战。 Flink 1.8.0 版本通过添加对过期状态对象的持续后台清理的支持，显着改进了状态 TTL 功能。新的清理机制使您无需手动实施状态清理。由于他们的懒惰本性，他们也更有效率。状态 TTL 使您可以控制应用程序状态的大小，以便您可以专注于应用程序的核心逻辑。