徐州建设网站价格,亚马逊品牌备案的网站怎么做,wordpress博客 手机网页 wap,企业网站源码 html5+xmlFlink系列之#xff1a;Flink 1.8.0 中的状态 TTL#xff1a;如何在 Apache Flink 中自动清理应用程序状态 一、状态的瞬态性质二、用于持续清理应用程序状态的状态 TTL三、倒垃圾四、保持完整状态快照干净五、堆状态后端的增量清理六、RocksDB 后台压缩以过滤掉过期状态七、… Flink系列之Flink 1.8.0 中的状态 TTL如何在 Apache Flink 中自动清理应用程序状态 一、状态的瞬态性质二、用于持续清理应用程序状态的状态 TTL三、倒垃圾四、保持完整状态快照干净五、堆状态后端的增量清理六、RocksDB 后台压缩以过滤掉过期状态七、使用计时器进行急切的状态清理八、未来的工作九、总结 许多有状态流应用程序的常见要求是自动清理应用程序状态以有效管理状态大小或控制应用程序状态可以访问的时间例如由于 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();ValueStateDescriptorLong lastUserLogin new ValueStateDescriptor(lastUserLogin, Long.class);lastUserLogin.enableTimeToLive(ttlConfig);这段代码使用Apache Flink提供的StateTtlConfig来设置状态的TTLTime-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 使您可以控制应用程序状态的大小以便您可以专注于应用程序的核心逻辑。
