Flink 1.13 版本 在某些场景下 Flink 用户状态一直在无限增长，一些用例需要能够自动清理旧的状态。...Checkpoint)时清理过期状态，这样可以大大减小快照存储，但需要注意的是本地状态中过期数据并不会被清理。...第二个参数定义了在每次处理记录时是否额外触发清理。堆状态后端的默认后台清理每次触发检查 5 个条目，处理记录时不会额外进行过期数据清理。...目前仅堆状态后端实现了增量清理。为 RocksDB 状态后端设置增量清理不会有任何效果。 如果堆状态后端与同步快照一起使用，全局迭代器在迭代时保留所有 Key 的副本，因为它的特定实现不支持并发修改。...如果使用 RocksDB StateBackend，则会调用 Flink 指定的压缩过滤器进行后台清理。RocksDB 周期性运行异步压缩来合并状态更新并减少存储。
在对应的状态后端（StateBackend）中，会调用对应的 create 方法获取到 StateDescriptor 中的值。在 Flink 中状态描述叫作 StateDescriptor。...执行检查点的时候，会把 State 的快照数据保存到 JobManager 进程的内存中。MemoryStateBackend 可以使用异步的方式进行快照，也可以使用同步的方式。...因为 sstable 是不可变的，Flink 对比前一个检查点创建和删除的 RocksDB sstable 文件就可以计算出状态有哪些改变。...调用 cleanupFullSnapshot() 做完整快照时清理后，在获取完整状态时激活清理。 调用 cleanupIncrementally 通过增量触发器渐进清理 State。...当进行状态访问或者处理数据时，在回调函数中进行清理。每次递增清理触发时，遍历 StateBackend 中的状态，清理过期的。
流处理应用程序通常是有状态的，“记住”已处理事件的信息，并使用它来影响进一步的事件处理。在Flink中，记忆的信息（即状态）被本地存储在配置的状态后端中。...为了防止发生故障时丢失数据，状态后端会定期将其内容快照保存到预先配置的持久性存储中。该RocksDB[1]状态后端（即RocksDBStateBackend）是Flink中的三个内置状态后端之一。...状态在Flink 为了更好地了解Flink中的状态和状态后端，区分运行中状态和状态快照非常重要。运行中状态（也称为工作状态）是Flink作业正在处理的状态。...将RocksDB视为需要在群集上运行并由专门的管理员进行管理的分布式数据库是一个常见的误解。RocksDB是用于快速存储的可嵌入持久化的kv存储。它通过Java本机接口（JNI）与Flink进行交互。...此外，运行中的状态也不要求高可用性。如果需要高磁盘吞吐量，则首选本地SSD磁盘。 状态快照将持久保存到远程持久性存储中。在状态快照期间，TaskManager会为运行中的状态拍摄快照并远程存储。
RocksDB状态后端为每个存储值、列表条目或映射条目添加8个字节; 目前只支持与处理时间相关的TTLs; 如果试图使用启用TTL的描述符或使用启用TTL的描述符恢复先前在没有TTL的情况下配置的状态，...此外，您可以在获取完整状态快照时激活清理操作，这将减少其大小。 在当前实现下，本地状态不会被清除，但在从前一个快照恢复时，它不会包含已删除的过期状态。可以在StateTtlConfig 中配置。...； 如果使用堆状态后端进行同步快照，全局迭代器在跌倒时会保留所有键的副本，因为它的特性不支持对并发数的修改。...使用此功能将增加内存消耗。异步快照进行对状态的保存就没有这种情况发生； 对于现有的作业，可以通过在StateTtlConfig中设置这种清理策略能够随时被激活和停用，例如：从保存点重新启动后。...3、Cleanup during RocksDB compaction 如果使用RocksDB进行状态的管理，另一个清理策略就是激活Flink的压缩过滤这个策略。
当进行checkpoint时，这个状态后端会对当前的状态进行快照，并且将其作为checkpoint ACK消息的一部分发送给JobManager(master)，该JobManager将其存储在它的堆上...: file:///data/flink/checkpoints FsStateBackend将流计算数据状态存储在TaskManager的内存中，在数据流遇到检查点屏障时，再将数据快照存储在配置好的文件系统中...在checkpoint时，此后端会将状态快照写入配置的文件系统和目录的文件中，同时会在JobManager的内存中（在高可用场景下会存在 Zookeeper 中）存储极少的元数据。...在 checkpoint 时，整个 RocksDB 数据库会被存储到配置的文件系统中，或者在超大状态作业时可以将增量的数据存储到配置的文件系统中。...如果你希望为你的集群中的所有作业创建一个非默认的状态后端，你可以通过在flink-conf.yaml中指定一个新的默认后端。默认的状态后端可以在每个作业的基础上进行覆盖，如下所示。
全量快照时进行清理 另外，你可以启用全量快照时进行清理的策略，这可以减少整个快照的大小。当前实现中不会清理本地的状态，但从上次快照恢复时，不会恢复那些已经删除的过期数据。...增量数据清理 另外可以选择增量式清理状态数据，在状态访问或/和处理时进行。如果某个状态开启了该清理策略，则会在存储后端保留一个所有状态的惰性全局迭代器。...在 RocksDB 压缩时清理 如果使用 RocksDB state backend，则会启用 Flink 为 RocksDB 定制的压缩过滤器。...RocksDB 会周期性的对数据进行合并压缩从而减少存储空间。Flink 提供的 RocksDB 压缩过滤器会在压缩时过滤掉已经过期的状态数据。...时间戳更新的越频繁，状态的清理越及时，但由于压缩会有调用 JNI 的开销，因此会影响整体的压缩性能。RocksDB backend 的默认后台清理策略会每处理 1000 条数据进行一次。
另外需要牢记的是从状态中获取的值取决于输入元素所代表的 key。 因此，在不同 key 上调用同一个接口，可能得到不同的值。 你必须创建一个 StateDescriptor，才能得到对应的状态句柄。...增量数据清理 # 另外可以选择增量式清理状态数据，在状态访问或/和处理时进行。如果某个状态开启了该清理策略，则会在存储后端保留一个所有状态的惰性全局迭代器。...在 RocksDB 压缩时清理 如果使用 RocksDB state backend，则会启用 Flink 为 RocksDB 定制的压缩过滤器。...RocksDB 会周期性的对数据进行合并压缩从而减少存储空间。 Flink 提供的 RocksDB 压缩过滤器会在压缩时过滤掉已经过期的状态数据。...对于元素序列化后长度不固定的列表状态，TTL 过滤器需要在每次 JNI 调用过程中，额外调用 Flink 的 java 序列化器， 从而确定下一个未过期数据的位置。
12、Flink 状态接口包括哪些？...在Flink中使用状态，包含两种状态接口：（1）状态操作接口：使用状态对象本身存储，写入、更新数据。（2）状态访问接口：从StateBackend获取状态对象本身。...在做持久化的过程中，状态可能会被持续修改， 基于内存的状态后端使用 CopyOnWriteStateTable 来保证线程安全，RocksDBStateBackend则使用RocksDB的快照机制，使用快照来保证线程安全...因为 sstable 是不可变的，Flink对比前一个检查点创建和删除的RocksDB sstable 文件就可以计算出状态有哪些发生改变。...TypeInformation 是 Flink 类型系统的核心类在Flink中，当数据需要进行序列化时，会使用TypeInformation的生成序列化器接口调用一个 createSerialize()
⭐ 怎么学习 Flink 中的状态、状态管理相关的概念呢？ ⭐ Flink 中状态的分类？ ⭐ Flink 中状态的使用方式？ ⭐ Flink 状态后端的分类及使用建议？...： ⭐ 状态：指 Flink 程序中的状态数据，博主认为也能代指用户在使用 DataStream API 编写程序来操作 State 的接口。...⭐ 状态后端：做状态数据（持久化，restore）的工具就叫做状态后端。比如你在 Flink 中见到的 RocksDB、FileSystem 的概念就是指状态后端。...这些状态后端就是实际存储上面的状态数据的。比如配置了 RocksDB 作为状态后端，MapState 的数据就会存储在 RocksDB 中。...⭐ 状态的创建方式：如果需要使用 operator-state，需要实现 CheckpointedFunction 或 ListCheckpointed 接口 ⭐ DataStream API 中，operator-state
或者使用类似于 RocksDB 这样的状态后端， RocksDB 会开辟堆外存储空间，但 IO 速度会变慢，需要权衡。...（详情可以点击文末原文链接查看：Flink 滑动窗口优化） 四，状态后端使用 RocksDB，还没有碰到被撑爆的问题 3、为什么用 Flink 问题：为什么使用 Flink 替代 Spark.../task的状态来生成快照，从而将这些状态数据定期持久化存储下来，当Flink程序一旦意外崩溃时，重新运行程序时可以有选择地从这些快照进行恢复，从而修正因为故障带来的程序数据异常。...Flink 会以 checkpoint 的形式对各个任务的 状态进行快照，用于保证故障恢复时的状态一致性。...解答：使用大容量的 Kafka 把数据先放到消息队列里面作为数据源，再使用 Flink 进行消费，不过这样会影响到一点实时性。
如果发生故障，Flink 会使用最近一个完成的快照来恢复应用程序。有些用户的作业状态达到 GB 甚至 TB 级别。...如何使用 目前只能在 RocksDB 状态后端上使用增量 Checkpoint，Flink 依赖 RocksDB 内部的备份机制来生成 Checkpoint 文件。...在这个基础上，Flink 会跟踪 RocksDB 自上一个 Checkpoint 以来创建和删除了哪些 sstable 文件，并且由于 sstable 是不可变的，所以 Flink 使用 sstable...当 Checkpoint 完成时，Flink 会在共享状态注册表中创建两条记录并将它们的计数设置为 1。...从Checkpoint恢复以及性能 开启增量 Checkpoint 之后，不需要再进行其他额外的配置就可以在发生故障时从状态中恢复。
本期文章，菌哥就带大家来总结一下，在面试过程中，Flink常被问到的知识点有哪些？...或者使用类似于 RocksDB 这样的状态后端， RocksDB 会开辟堆外存储空间，但 IO 速度会变慢，需要权衡。.../task的状态来生成快照，从而将这些状态数据定期持久化存储下来，当Flink程序一旦意外崩溃时，重新运行程序时可以有选择地从这些快照进行恢复，从而修正因为故障带来的程序数据异常。...Flink 会以 checkpoint 的形式对各个任务的 状态进行快照，用于保证故障恢复时的状态一致性。...解答：使用大容量的 Kafka 把数据先放到消息队列里面作为数据源，再使用 Flink 进行消费，不过这样会影响到一点实时性。
下面的几个场景都需要使用流处理的状态功能： 数据流中的数据有重复，我们想对重复数据去重，需要记录哪些数据已经流入过应用，当新数据流入时，根据已流入过的数据来判断去重。...Operator State KeyedState 是在进行 KeyBy 之后进行状态操作时使用的状态类型，那么像 Source、Sink算子是不会进行 KeyBy 操作的，当这类算子也需要用到状态，应该怎么操作呢...如果List State的基数较大时，不要使用这种方式的redistribution。因为容易引起OOM。 调用不同的获取状态对象的接口，会使用不同的状态分配算法。...主要区分为： restoreWithRescaling：从多个增量的状态后端恢复，需要进行扩缩容。在这个过程中会创建一个临时的RocksDB实例用于关key-groups。...临时RocksDB当中的数据在都会复制到实际使用的RocksDB的实例当中。 restoreWithoutRescaling：从单个远程的增量状态后端恢复，无需进行扩缩容。
状态后端 Flink提供了3种存储状态的方式： 内存 文件系统 RocksDB 这三种存储方式又被称为状态后端（State Backend）。...当选择使用文件系统作为后端时，正在计算的数据会被暂存在TaskManager的内存中。...Checkpoint时，此后端会将状态快照写入配置的文件系统中，同时会在JobManager的内存中或者在 Zookeeper 中（高可用情况）存储极少的元数据。...使用RocksDB作为后端时，Flink会将实时处理中的数据使用RocksDB存储在本地磁盘上。...Checkpoint时，整个RocksDB数据库会被存储到配置的文件系统中，同时Flink会将极少的元数据存储在JobManager的内存中，或者在Zookeeper中（高可用情况）。
RocksDBKeyedStateBackend增量快照介绍 这里介绍一下大家在大状态场景下经常需要调优的 RocksDBKeyedStateBackend 增量快照。...RocksDB 具有 append-only 特性，Flink 利用这一特性将两次 checkpoint 之间 SST 文件列表的差异作为状态增量上传到分布式文件系统上，并通过 JobMaster 中的...RocksDb大状态优化 截至当前，Flink 作业的状态后端仍然只有 Memory、FileSystem 和 RocksDB 三种可选，且 RocksDB 是 状态数据量较大（GB 到 TB 级别）...或代码中指定 new EmbededRocksDBStateBackend(true) 开启本地恢复 当flink任务失败时，可以基于本地的状态信息进行恢复任务。可能不需要从hdfs拉取数据。...RocksDB 相关参数在前面已说明，可以在 flink-conf.yaml 指定，也可以在 Job 的代码中调用 API 单独指定，这里不再列出。
在Flink 1.8.0中，该功能得到了扩展，包括对RocksDB和堆状态后端（FSStateBackend和MemoryStateBackend）的历史数据进行持续清理，从而实现旧条目的连续清理过程（...完整快照自动删除过期状态 当获取检查点或保存点的完整快照时，Flink 1.6.0已经支持自动删除过期状态。大家注意，过期状态删除不适用于增量检查点。...只有当用户从快照重新加载其状态到本地时，才会清除用户的本地状态。 由于上述这些限制，FLink应用程序仍需要在Flink 1.6.0中过期后主动删除状态。...RocksDB后台压缩可以过滤掉过期状态 如果你的Flink应用程序使用RocksDB作为状态后端存储，则可以启用另一个基于Flink特定压缩过滤器的清理策略。...RocksDB定期运行异步压缩以合并状态更新并减少存储。Flink压缩过滤器使用TTL检查状态条目的到期时间戳，并丢弃所有过期值。
当进行checkpoint时，它只写入字节序列到checkpoint中。Flink并不知道状态的数据结构，并且只能看到raw字节。...请使用AggregatingState和 AggregatingStateDescriptor替代。 首先需要记住的是这些状态对象只能用来与状态进行交互。...状态不一定存储在内存中，但是可能存储在磁盘或者其他地方。第二个需要记住的是，从状态获取的值依赖于输入元素的key。...因此如果包含不同的key，那么在你的用户函数中的一个调用获得的值和另一个调用获得值可能不同。 为了获得状态句柄，必须创建一个StateDescriptor。...，也即是状态后端不能是RocksDB。
当检查点的完成时间超过检查点间隔时，在进行中的检查点完成之前不会触发下一个检查点。 默认情况下，一旦正在进行的检查点完成，将立即触发下一个检查点。...当手动触发保存点时，它可能与正在进行的检查点同时进行。 RocksDB调优 许多大型 Flink 流应用程序的状态存储主力是 RocksDB 状态后端。...默认情况下，RocksDB 状态后端使用 Flink 为 RocksDB 缓冲区和缓存管理的内存预算（state.backend.rocksdb.memory.managed: true）。...本地任务恢复 动机 在 Flink 的检查点中，每个任务都会生成其状态的快照，然后将其写入分布式存储。 每个任务通过发送描述状态在分布式存储中的位置的句柄来向Jobmanager确认状态的成功写入。...对于增量快照，本地状态基于 RocksDB 的原生检查点机制。这种机制也被用作创建主副本的第一步，这意味着在这种情况下，创建辅助副本不会引入额外的成本。
⭐ 状态：本质来说就是数据，在 Flink 中，其实就是 Flink 提供给用户的状态编程接口。比如 flink 中的 MapState，ValueState，ListState。...然后可以在 Flink 任务 failover 时，从远程把状态数据恢复到 Flink 任务中，保障数据质量。...，其声明了整个任务的状态管理后端类型； ⭐ 每个格子中的内容就是用户在配置 xx 状态后端（列）时，给用户使用的状态（行）生成的状态后端实例，生成的这个实例就是在 Flink 中实际用于管理用户使用的状态的组件...用户在配置 rocksdb 时，会使用 RocksdbKeyedStateBackend 去管理状态；用户在配置 memory，filesystem 时，会使用 HeapKeyedStateBackend...比如计算 DAU 这种大数据量去重，大状态的任务都建议直接使用 RocksDB 状态后端。 到生产环境中： ⭐ 如果状态很大，使用 Rocksdb；如果状态不大，使用 Filesystem。
领取专属 10元无门槛券
手把手带您无忧上云
云服务器
即时通信 IM
ICP备案
对象存储
实时音视频
