Flink系列之：State Time-To-Live (TTL)

一、TTL

• Flink的TTL（Time-To-Live）是一种数据过期策略，用于指定数据在流处理中的存活时间。TTL可以应用于Flink中的状态或事件时间窗口，以控制数据的保留时间。
• 当应用程序使用状态进行计算时，状态可能会消耗存储资源。TTL可以用来设置状态的最大生存时间，超过该时间的状态将被自动清理，以释放存储资源。这可以帮助应对状态数据的增长和资源限制问题。
• 对于事件时间窗口，TTL可以用来指定窗口的持续时间。当到达窗口结束时间后，该窗口的结果将被输出，并且窗口中的所有数据将被清理。这可以确保计算结果及时输出，并释放计算资源。
• 通过设置适当的TTL值，可以控制数据的保留时间，避免资源浪费和计算延迟。TTL的使用可以根据具体应用场景和需求进行配置，以实现数据管理的灵活性和效率。
• 可以将生存时间 (TTL) 分配给任何类型的键控状态。如果配置了 TTL 并且状态值已过期，则将尽最大努力清除存储的值，这将在下面更详细地讨论。
• 所有状态集合类型都支持每条目 TTL。这意味着列表元素和映射条目独立过期。
• 为了使用状态 TTL，必须首先构建一个 StateTtlConfig 配置对象。然后可以通过传递配置在任何状态描述符中启用 TTL 功能：

二、TTL实现代码

java代码：

import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.seconds(1)).setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired).build();ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("text state", String.class);
stateDescriptor.enableTimeToLive(ttlConfig);

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

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

Scala代码：

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

Python代码：

from pyflink.common.time import Time
from pyflink.common.typeinfo import Types
from pyflink.datastream.state import ValueStateDescriptor, StateTtlConfigttl_config = StateTtlConfig \.new_builder(Time.seconds(1)) \.set_update_type(StateTtlConfig.UpdateType.OnCreateAndWrite) \.set_state_visibility(StateTtlConfig.StateVisibility.NeverReturnExpired) \.build()state_descriptor = ValueStateDescriptor("text state", Types.STRING())
state_descriptor.enable_time_to_live(ttl_config)

该配置有几个选项需要考虑：

• newBuilder方法的第一个参数是必需的，它是生存时间值。
• 更新类型配置何时刷新状态 TTL（默认为 OnCreateAndWrite）：
  • StateTtlConfig.UpdateType.OnCreateAndWrite - 仅在创建和写入访问时
  • StateTtlConfig.UpdateType.OnReadAndWrite - 也用于读取访问
  • （注：如果同时将状态可见性设置为StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp，状态读缓存将被禁用，这会导致PyFlink中的一些性能损失）

状态可见性配置如果尚未清除过期值，是否在读取访问时返回过期值（默认为 NeverReturnExpired）：

• StateTtlConfig.StateVisibility.NeverReturnExpired - 永远不会返回过期值

（注：状态读/写缓存将被禁用，这会导致 PyFlink 中的一些性能损失）

• StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp - 如果仍然可用则返回

在 NeverReturnExpired 的情况下，过期状态的行为就好像它不再存在一样，即使它仍然需要被删除。该选项对于数据必须在 TTL 之后严格无法进行读取访问的用例很有用，例如处理隐私敏感数据的应用程序。

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

笔记：

• 状态后端存储上次修改的时间戳以及用户值，这意味着启用此功能会增加状态存储的消耗。堆状态后端存储一个附加的 Java 对象，其中包含对用户状态对象的引用和内存中的原始 long 值。 RocksDB 状态后端为每个存储值、列表条目或映射条目添加 8 个字节。
• 当前仅支持涉及处理时间的 TTL。
• 尝试使用启用 TTL 的描述符恢复之前未配置 TTL 的状态，反之亦然，将导致兼容性失败和 StateMigrationException。
• TTL 配置不是检查点或保存点的一部分，而是 Flink 在当前运行的作业中处理它的一种方式。
• 不建议通过将 ttl 从短值调整为长值来恢复检查点状态，这可能会导致潜在的数据错误。
• 目前，仅当用户值序列化程序可以处理空值时，具有 TTL 的映射状态才支持空用户值。如果序列化器不支持 null 值，则可以使用 NullableSerializer 对其进行包装，但需要在序列化形式中增加一个额外字节。
• 启用 TTL 的配置后，StateDescriptor 中的 defaultValue 实际上已被弃用，将不再生效。这样做的目的是使语义更加清晰，并让用户在状态内容为空或过期时手动管理默认值。

三、过期状态的清理

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

Java代码：

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

Scala代码：

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

Python代码：

from pyflink.common.time import Time
from pyflink.datastream.state import StateTtlConfigttl_config = StateTtlConfig \.new_builder(Time.seconds(1)) \.cleanup_in_rocksdb_compact_filter(1000, Time.hours(1)) \.build()

为了对后台的一些特殊清理进行更细粒度的控制，您可以如下所述单独配置它。目前，堆状态后端依赖于增量清理，RocksDB 后端使用压缩过滤器进行后台清理。

完整快照中的清理

此外，您可以在拍摄完整状态快照时激活清理，这将减少其大小。在当前实现下，本地状态不会被清除，但在从以前的快照恢复时，它不会包括删除的过期状态。可以在StateTtlConfig中配置：

Java代码：

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

Scala代码：

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

Python代码：

from pyflink.common.time import Time
from pyflink.datastream.state import StateTtlConfigttl_config = StateTtlConfig \.new_builder(Time.seconds(1)) \.cleanup_full_snapshot() \.build()

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

对于现有作业，可以在 StateTtlConfig 中随时激活或停用此清理策略，例如从保存点重新启动后。

增量清理

另一种选择是逐步触发某些状态条目的清理。触发器可以是来自每个状态访问或/和每个记录处理的回调。如果此清理策略对于某些状态是活动的，则存储后端会在其所有条目上为此状态保留一个惰性全局迭代器。每次触发增量清理时，迭代器都会前进。检查遍历的状态条目并清除过期的状态条目。

该功能可以在 StateTtlConfig 中配置：

Java代码：

import org.apache.flink.api.common.state.StateTtlConfig;StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.seconds(1)).cleanupIncrementally(10, true).build();

Scala代码：

import org.apache.flink.api.common.state.StateTtlConfig
val ttlConfig = StateTtlConfig.newBuilder(Time.seconds(1)).cleanupIncrementally(10, true).build

Python：

from pyflink.common.time import Time
from pyflink.datastream.state import StateTtlConfigttl_config = StateTtlConfig \.new_builder(Time.seconds(1)) \.cleanup_incrementally(10, True) \.build()

该策略有两个参数。第一个是每次清理触发的检查状态条目数。它总是在每次状态访问时触发。第二个参数定义是否在每次记录处理时额外触发清理。堆后端的默认后台清理会检查 5 个条目，而不会针对每个记录处理进行清理。

笔记：

• 如果没有对状态进行访问或没有处理任何记录，则过期状态将持续存在。
• 增量清理所花费的时间会增加记录处理延迟。
• 目前增量清理仅针对堆状态后端实现。对 RocksDB 设置它不会有任何效果。
• 如果堆状态后端与同步快照一起使用，则全局迭代器在迭代时会保留所有键的副本，因为其特定实现不支持并发修改。启用此功能将增加内存消耗。异步快照则不存在这个问题。
• 对于现有作业，可以在 StateTtlConfig 中随时激活或停用此清理策略，例如从保存点重新启动后。

RocksDB 压缩期间的清理

如果使用 RocksDB 状态后端，将调用 Flink 特定的压缩过滤器进行后台清理。 RocksDB 定期运行异步压缩来合并状态更新并减少存储。 Flink 压缩过滤器使用 TTL 检查状态条目的过期时间戳并排除过期值。

该功能可以在 StateTtlConfig 中配置：

Java代码：

import org.apache.flink.api.common.state.StateTtlConfig;StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.seconds(1)).cleanupInRocksdbCompactFilter(1000, Time.hours(1)).build();

Scala代码：

import org.apache.flink.api.common.state.StateTtlConfigval ttlConfig = StateTtlConfig.newBuilder(Time.seconds(1)).cleanupInRocksdbCompactFilter(1000, Time.hours(1)).build

Python代码：

from pyflink.common.time import Time
from pyflink.datastream.state import StateTtlConfigttl_config = StateTtlConfig \.new_builder(Time.seconds(1)) \.cleanup_in_rocksdb_compact_filter(1000, Time.hours(1)) \.build()

RocksDB 压缩过滤器每次处理一定数量的状态条目后都会从 Flink 查询当前时间戳，用于检查过期情况。您可以更改它并将自定义值传递给 StateTtlConfig.newBuilder(…).cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries) 方法。更频繁地更新时间戳可以提高清理速度，但会降低压缩性能，因为它使用来自本机代码的 JNI 调用。 RocksDB 后端的默认后台清理会在每次处理 1000 个条目时查询当前时间戳。

定期压缩可以加快过期状态条目的清理速度，特别是对于很少访问的状态条目。早于该值的文件将被拾取进行压缩，并重新写入到与之前相同的级别。它确保文件定期通过压缩过滤器。您可以更改它并将自定义值传递给 StateTtlConfig.newBuilder(…).cleanupInRocksdbCompactFilter(long queryTimeAfterNumEntries, Time periodicalCompactionTime) 方法。定期压缩秒数的默认值为 30 天。您可以将其设置为 0 以关闭定期压缩，或设置一个较小的值以加速过期状态条目清理，但它会触发更多压缩。

您可以通过激活 FlinkCompactionFilter 的调试级别来从 RocksDB 过滤器的本机代码激活调试日志：

log4j.logger.org.rocksdb.FlinkCompactionFilter=DEBUG

笔记：

• 在压缩过程中调用 TTL 过滤器会减慢速度。 TTL 过滤器必须解析上次访问的时间戳，并检查正在压缩的每个键的每个存储状态条目的过期时间。如果是集合状态类型（列表或映射），还会针对每个存储的元素调用检查。
• 如果此功能与包含非固定字节长度元素的列表状态一起使用，则本机 TTL 过滤器必须在每个状态条目（其中至少第一个元素已过期）额外通过 JNI 调用该元素的 Flink java 类型序列化器确定下一个未过期元素的偏移量。
• 对于现有作业，可以在 StateTtlConfig 中随时激活或停用此清理策略，例如从保存点重新启动后。
• 周期性压缩仅在启用 TTL 时才起作用。