Flink状态与容错【状态生存时间TTL，清理过期状态】

发布时间：2024-11-23 14:56

避免在疲劳状态下驾驶，保持清醒状态。 #生活技巧# #驾驶安全技巧# #车辆保养与安全#

一.状态生存时间（TTL）

任何类型的键控状态都可以配置状态的生存期（TTL）。如果配置了TTL且状态值已过期，则将尽力清除存储的值。

所有状态收集类型均支持按条目TTL。这意味着列表元素和映射条目独立过期。

为了使用状态TTL，必须首先构建一个StateTtlConfig配置对象。然后可以通过传递配置在任何状态描述符中启用TTL功能：

import org.apache.flink.api.common.state.StateTtlConfig import org.apache.flink.api.common.state.ValueStateDescriptor import org.apache.flink.api.common.time.Time val ttlConfig = StateTtlConfig .newBuilder(Time.seconds(1)) .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) .build val stateDescriptor = new ValueStateDescriptor[String]("text state", classOf[String]) stateDescriptor.enableTimeToLive(ttlConfig) 123456789101112

该配置有几个选项可供考虑：

该newBuilder方法的第一个参数是强制性的，它是生存时间值。

更新类型配置何时刷新状态TTL（默认为OnCreateAndWrite）：

StateTtlConfig.UpdateType.OnCreateAndWrite -仅在创建和写访问权限时 StateTtlConfig.UpdateType.OnReadAndWrite -也具有读取权限

状态可见性用于配置是否清除尚未过期的默认值（默认情况下NeverReturnExpired）：

StateTtlConfig.StateVisibility.NeverReturnExpired -永不返回过期值 StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp -如果仍然可用，则返回

在NeverReturnExpired的情况下，即使仍必须删除过期状态，其行为也好像不再存在一样。对于严格在TTL之后数据必须变得不可用于读取访问的用例，例如使用隐私敏感数据的应用程序，该选项很有用。

另一个选项ReturnExpiredIfNotCleanedUp允许在清理之前返回过期状态。

笔记：

状态后端与用户值一起存储上次修改的时间戳，这意味着启用此功能会增加状态存储的消耗。堆状态后端将附加的Java对象与对用户状态对象的引用以及原始的long值一起存储在内存中。RocksDB状态后端为每个存储的值，列表条目或映射条目添加8个字节。当前仅支持有关处理时间的 TTL 。尝试使用启用了TTL的描述符恢复状态（以前没有配置TTL），反之亦然，这将导致兼容性失败和StateMigrationException。 TTL配置不是检查点或保存点的一部分，而是Flink在当前正在运行的作业中如何对待它的一种方式。仅当用户值序列化程序可以处理空值时，带有TTL的映射状态当前才支持空用户值。如果序列化器不支持空值，则可以用NullableSerializer序列化形式包装一个额外的字节来包装它。

二.清理过期状态

默认情况下，过期的值会在读取时显式删除，例如ValueState#value，并且如果配置的状态后端支持，则会定期在后台收集垃圾。可以在以下位置禁用后台清理StateTtlConfig：

import org.apache.flink.api.common.state.StateTtlConfig val ttlConfig = StateTtlConfig .newBuilder(Time.seconds(1)) .disableCleanupInBackground .build 12345

要对后台的某些特殊清理进行更细粒度的控制，可以按如下所述分别进行配置。当前，堆状态后端依赖于增量清理，而RocksDB后端使用压缩过滤器进行后台清理。

三.完整快照中的清理

此外，可以在拍摄完整状态快照时激活清除操作，这将减小其大小。在当前实现下不会清除本地状态，但是如果从先前的快照还原，则不会包括已删除的过期状态。可以在StateTtlConfig以下位置进行配置：

import org.apache.flink.api.common.state.StateTtlConfig import org.apache.flink.api.common.time.Time val ttlConfig = StateTtlConfig .newBuilder(Time.seconds(1)) .cleanupFullSnapshot .build 1234567

此选项不适用于RocksDB状态后端中的增量检查点。

笔记：

对于现有作业，可以在StateTtlConfig中的任何时间（例如，从保存点重新启动后）激活或取消激活此清除策略。