Kafka Connector 是在Flink中使用算子状态的一个很好的例子。Kafka消费者的每个并行实例都要维护一个topic分区和偏移量的map作为其Operator State。...Flink对状态的数据结构一无所知，只能看到原始字节。 所有数据流函数都可以使用Managed State，但Raw State接口只能在实现算子时使用。...建议使用Managed State（而不是Raw State），因为在Managed State下，Flink可以在并行度发生变化时自动重新分配状态，并且还可以更好地进行内存管理。...这意味着这种类型的状态只能用于KeyedStream，可以通过stream.keyBy（...）创建。 现在，我们先看看可用状态的不同类型，然后我们会看到如何在程序中使用。...注意一下状态是如何被初始化，类似于keyed state状态，使用包含状态名称和状态值类型相关信息的StateDescriptor： Java版本: ListStateDescriptor<Tuple2
在本节中，您将了解Flink为编写有状态程序提供的api。请参阅有状态流处理以了解有状态流处理背后的概念。...使用 Keyed State keyed state 接口提供不同类型状态的访问接口，这些状态都作用于当前输入数据的 key 下。...接下来，我们会介绍不同类型的状态，然后介绍如何使用他们。...在 RocksDB 压缩时清理 如果使用 RocksDB state backend，则会启用 Flink 为 RocksDB 定制的压缩过滤器。...Flink 提供的 RocksDB 压缩过滤器会在压缩时过滤掉已经过期的状态数据。
下面我们会介绍这个新的状态 TTL 功能的动机并讨论其用例。此外，我们还会展示如何使用和配置它，以及解释 Flink 如何使用 TTL 管理内部状态。文章最后还展望了对未来的改进和扩展。 1....Flink有状态流处理 任何实时流应用程序都会包含有状态操作。Flink 为容错状态流处理提供了许多强大的功能。...Apache Flink 1.6.0 版本开始引入了状态 TTL 功能。流处理应用的开发者可以将算子的状态配置为在一定时间内没有被使用下自动过期。过期状态稍后由惰性清理策略进行垃圾收集。...以下 Java 示例展示了如何创建状态 TTL 配置并将其提供给状态描述符，该描述符将用户的上次登录时间作为 Long 值保存： import org.apache.flink.api.common.state.StateTtlConfig...由于这种惰性删除方式，永远不会再次访问的过期状态将永远占用存储空间，除非它被垃圾回收。 如果应用程序逻辑没有明确的处理，那么如何删除过期状态呢？一般来说，有不同的策略可以在后台进行删除。
Hi~朋友，关注置顶防止错过消息 什么是有状态的计算 使用状态的场景 为什么需要状态管理 理想状态管理的特点 Flink状态分类 Managed State分类 Keyed Stated特点 Operator...State特点 Keyed Stated的具体分类 如何保存状态 Checkpoint和Savepoint区别 状态保存在哪里 什么是有状态的计算？...可靠：状态需要可以被持久化，保证宕机后可以恢复 Flink状态分类 Managed State RawState 状态管理方式 Flink Runtime自动管理：自动存储、自动恢复、内存优化 用户自己管理...，需要自己序列化 状态数据结构 已知的数据结构：Value、List、Map等 字节数组byte[] 推荐使用场景 大多数情况下可以使用 自定义Operator时使用 Managed State分类...AggregatingState：AggregatingState和ReducingState的区别是在访问接口，Reducing的add和get的元素都是同一个类型，但是Aggregating输入的是IN，输出的是OUT 如何保存状态
概念 在Flink架构体系中，有状态计算可以说是Flink非常重要的特性之一 Flink优势： 支持高吞吐、低延迟、高性能 支持事件时间Event_time概念 支持有状态计算 有状态计算是指: 在程序计算过程中...，在Flink程序内部存储计算产生的中间结果，并提供给后续Function或算子计算结果使用。...最大值、均值等聚合指标（如pv,uv）: 需要利用状态来维护当前计算过程中产生的结果，例如事件的总数、总和以及最大，最小值等 机器学习场景，维护当前版本模型使用的参数 其他需要使用历史数据的计算...在Flink中推荐用户使用Managed State管理状态数据，主要原因是Managed State能够更好地支持状态数据的重平衡以及更加完善的内存管理。...此时需要给用户发送一个信息提醒用户，提高支付转换率 需求分析 本需求可以使用CEP来实现， 这里推荐使用process function原生的状态编程。
即你可以将中间的计算结果进行保存，并提供给后续的计算使用： 具体而言，Flink 又将状态 (State) 分为 Keyed State 与 Operator State： 2.1 算子状态 算子状态...，即假设算子的并行度是 2，那么其应有两个对应的算子状态： 2.2 键控状态 键控状态 (Keyed State) ：是一种特殊的算子状态，即状态是根据 key 值进行区分的，Flink 会为每类键值维护一个状态实例...如下图所示，每个颜色代表不同 key 值，对应四个不同的状态实例。需要注意的是键控状态只能在 KeyedStream 上进行使用，我们可以通过 stream.keyBy(...)...二、状态编程 2.1 键控状态 Flink 提供了以下数据格式来管理和存储键控状态 (Keyed State)： ValueState：存储单值类型的状态。...这里之所以要达到一定次数，是因为由于偶发原因，偶尔一次超过阈值并不能代表什么，故需要达到一定次数后才触发报警，这就需要使用到 Flink 的状态编程。
低阶Join 要在两个输入上实现低阶操作，应用程序可以使用 CoProcessFunction。...实现低阶 Join 通常遵循以下模式： 为一个输入（或两个）创建状态对象。 在从输入中收到元素时更新状态。 在从其他输入收到元素时扫描状态对象并生成 Join 结果。...在这里使用 KeyedProcessFunction 只是用来说明它的基本模式。...Flink同步调用 onTimer() 和 processElement() 方法。因此，用户不必担心状态的并发修改。 5.1 容错 定时器具有容错能力，并且与应用程序的状态一起进行快照。...Flink版本:1.8 英译对照: 事件: Events 状态: State 定时器: Timers 原文:Process Function (Low-level Operations)
Flink主要有两种基础类型的状态：keyed state 和operator state。...Flink并不知道状态的数据结构，并且只能看到raw字节。 所有的数据流函数都可以使用managed state，但是raw state接口只可以在操作算子的实现类中使用。...推荐使用managed state(而不是raw state)，因为使用managed state，当并行度变化时，Flink可以自动的重新分布状态，也可以做更好的内存管理。...这意味着这种类型的状态只能在KeyedStream中使用，它可以通过stream.keyBy(...)创建。 现在，我们首先看下不同类型的状态，然后展示如何在程序中使用它们。...TTL的使用也很简单，可以参考如下代码： import org.apache.flink.api.common.state.StateTtlConfig;import org.apache.flink.api.common.state.ValueStateDescriptor
Flink Metric内置的监控涵盖面丰富，支持： 1 系统级别的监控：CPU状态信息、内存状态信息等（默认关闭，需要更改配置文件打开，且lib目录下需要添加相关依赖jar包） 2 JVM级别的监控：...CPU占用率、堆内存使用情况、线程数量、GC已经类加载数量。...具体可以参考https://ci.apache.org/projects/flink/flink-docs-stable/monitoring/metrics.html 监控和使用 收集监控数据， 这里无论是采取...在实际使用建议使用Prometheus来收集Metrics，Prometheus可以较好的和infexdb和Gafana结合，完成数据收集和报表展示，而且还可以扩展Prometheus的告警模块进行下一步的告警...报表使用 这里就不用多说了，使用Prometheus的话，结合Grafana可以做到一站式的报表开发。
摘要本文将从状态的概念入手，详细介绍 Flink 中的状态分类、状态的使用、持久化及状态后端的配置。...一、Flink状态概念 Flink的处理机制核心：有状态的流式计算，那么什么是有状态，什么是无状态呢？...四、状态具体使用demo import dto.SensorReadingDTO; import org.apache.commons.lang3.StringUtils; import org.apache.flink.api.common.functions.MapFunction...; import org.apache.flink.util.Collector; import util.DateUtil; //如果传感器的温度差大于10度就预警 //使用状态记录上一次的状态 public...1、MemoryStateBackend 内存级的状态后端，会将键控状态作为内存中的对象进行管理，将它们存储在TaskManager的JVM堆上，而将checkpoint存储在JobManager的内存中
这篇文章我们将深入探讨有状态流处理，更确切地说是 Flink 中可用的不同状态后端。在以下部分，我们将介绍 Flink 的3个状态后端，它们的局限性以及根据具体案例需求选择最合适的状态后端。...在有状态的流处理中，当开发人员启用了 Flink 中的检查点功能时，状态会持久化存储以防止数据的丢失并确保发生故障时能够完全恢复。为应用程序选择何种状态后端，取决于状态持久化的方式和位置。...什么时候使用 MemoryStateBackend： 本地开发或调试时建议使用 MemoryStateBackend，因为这种场景的状态大小的是有限的。.../checkpoints s3://flink/checkpoints RocksDBStateBackend 将正在处理的数据使用 RocksDB 存储在本地磁盘上。...何时使用 RocksDBStateBackend： RocksDBStateBackend 非常适合处理大状态，长窗口，或大键值状态的有状态流处理作业。
这可能不适用于所有用例，但如果您的 Pipeline 必须维护内部状态（可能是进行一些聚合），则最好使状态可用于查询。 我们首先看看当我们使状态可查询以及何时查询时，在 Flink 内部的整体步骤。...下图显示了 Flink 内部发生的事情： image.png 我希望这个图是不言自明的，但总而言之，一旦提交了 Job，JobManager 就会从 JobGraph 构建 ExecutionGraph...然后 JobManager actor 会收到有关状态注册的通知，JobManager 将位置信息存储在 KvStateLocationRegistry 中，后面就可以在查询期间使用。 2....客户端向 JobManager actor 发送 KvStateLookup 消息，该请求应包含构建作业时使用的 JobId 以及状态名称。...然后客户端打开与 KvStateServer 的连接并使用 KvStateID 从注册表中获取状态。检索到状态后，将提交异步查询以从给定键的状态中获取值。得到的结果被序列化并发回客户端。
所以，Flink 在框架层面提供了状态的 Api，业务如果需要使用状态，直接使用框架提供的状态 api 来存储状态即可，至于如何存储的细节对于开发者来说是透明的，开发者专注自己的业务即可。...二、状态和容错的关系 Flink 在框架层面提供了算子状态（Operator State）和键控状态（Keyed State）。 算子状态是绑定在算子上的，而键控状态是绑定在某个key上的。...如何理解绑定在算子还是绑定在 key ？ 从概念上，每个算子的每个并行实例有一个算子状态。...从 Api 层面上，要使用算子状态，要实现 CheckpointedFunction，然后在初始化方法中，利用 context 拿到存储器再使用；而键控状态，可以直接在所有的 RichXXXFunction...那么键控状态，是跟某条数据绑定，和业务有直接关系，使用者自己来控制每条数据要存储什么样的状态。
如果状态过期，还会根据可见性参数，来决定是否返回已过期但还未清理的状态等等。状态的清理并不是即时的，而是使用了一种 Lazy 的算法来实现，从而减少状态清理对性能的影响。...StateVisibility：表示对已过期但还未被清理掉的状态如何处理，也是 Enum 对象。...这样在今后的 Flink 状态调用过程中，只要调用了状态的 get / put / update 等通用方法，都会自动地对失效状态进行判断、清理等操作，而 Flink 并不需要知道其背后的实现逻辑，只是把这些状态对象当作普通的来使用即可...这种封装的方式也体现了 Flink 的可扩展性，避免实现细节对上层调用逻辑产生干扰。 接下来，我们简单看下 Flink 是如何在 RocksDB 中实现 State TTL 的。...因此对于 Table / SQL 作业，两种机制可以结合使用，以应对逐渐增加的状态带来的挑战。
概述 Flink 提供了不同的状态终端，可以指定状态的存储方式和位置。 状态可以存储在Java的堆内或堆外。...根据你的状态终端，Flink 也可以管理应用程序的状态，这意味着 Flink 可以处理内存管理（可能会溢出到磁盘，如果有必要），以允许应用程序存储非常大的状态。...可用的状态终端 开箱即用，Flink 内置了如下状态终端： MemoryStateBackend FsStateBackend RocksDBStateBackend 如果没有配置，系统默认使用MemoryStateBackend...FsStateBackend 默认使用异步快照，以避免在写入状态检查点时阻塞处理管道。.../checkpoints")) 3.2 设置默认状态终端 可以使用配置键 state.backend 在 flink-conf.yaml 配置文件中配置默认状态终端。
Storm需要自己实现有状态的计算，比如借助于自定义的内存变量或者redis等系统，保证低延迟的情况下自己去判断实现有状态的计算，但是Flink就不需要这样，而且作为新一代的流处理系统，Flink非常重视...因此，用户经常不得不使用两个流处理框架 （一个用来保证 exactly-once，另一个用来对每个元素做低延迟处理），结果使基础设施更加复杂。 但是，Flink解决了这种问题。...Flink 检查点的核心作用是确保状态正确，即使遇到程序中断，也要正确。 记住这一基本点之后，我们用一个例子来看检查点是如何运行的。Flink 为 用户提供了用来定义状态的工具。...使用保存点更新Flink 应用程序的版本。新版本可以从旧版本生成的一个 保存点处开始执行. 端到端的一致性 ?...输入数据来自Kafka，在将状态内容传送到输出存储系统的过程中，如何保证 exactly-once 呢？这 叫作端到端的一致性。
Storm需要自己实现有状态的计算，比如借助于自定义的内存变量或者redis等系统，保证低延迟的情况下自己去判断实现有状态的计算，但是Flink就不需要这样，而且作为新一代的流处理系统，Flink非常重视...因此，用户经常不得不使用两个流处理框架 （一个用来保证 exactly-once，另一个用来对每个元素做低延迟处理），结果使基础设施更加复杂。 但是，Flink解决了这种问题。...Flink 检查点的核心作用是确保状态正确，即使遇到程序中断，也要正确。记住这一基本点之后，我们用一个例子来看检查点是如何运行的。Flink 为 用户提供了用来定义状态的工具。...使用保存点更新Flink 应用程序的版本。新版本可以从旧版本生成的一个 保存点处开始执行....输入数据来自Kafka，在将状态内容传送到输出存储系统的过程中，如何保证 exactly-once 呢？这 叫作端到端的一致性。
Flink 需要了解状态，以便使用检查点进行状态容错，并允许流应用程序使用保存点。 对状态进行了解有助于你对 Flink 应用程序进行扩展，这意味着 Flink 负责在并行实例之间进行重新分配状态。...Flink 的可查询状态queryable state功能允许你在 Flink 运行时在外部访问状态。 在使用状态时，阅读有关Flink的 State Backends 应该对你很有帮助。...Flink 提供不同的 State Backends，并指定状态的存储方式和位置。状态可以位于Java的堆内或堆外。...根据你的 State Backends，Flink也可以管理应用程序的状态，这意味着Flink进行内存管理(可能会溢写到磁盘，如果有必要)，以允许应用程序保持非常大的状态。...下一步 使用状态：显示如何在Flink应用程序中使用状态，并解释不同类型的状态。 检查点：描述如何启用和配置容错检查点。 可查询状态：解释如何在Flink运行时从外部访问状态。
InternalKVState 提供了只对 Flink 引擎暴露的接口比如 namespace set/get、val get、namespace merging，这些接口并不稳定，Flink 引擎希望对上层应用屏蔽...，而通过 State 接口暴露出来的方法普通用户是可以使用的。...ValueState：即类型为T的单值状态。这个状态与对应的key绑定，是最简单的状态了。它可以通过update方法更新状态值，通过value()方法获取状态值。...FoldingState：跟ReducingState有点类似，不过它的状态值类型可以与add方法中传入的元素类型不同（这种状态将会在Flink未来版本中被删除）。...Flink通过StateDescriptor来定义一个状态。这是一个抽象类，内部定义了状态名称、类型、序列化器等基础信息。
(raw state) 托管状态(managed state) 托管状态是由Flink框架管理的状态，而原始状态，由用户自行管理状态具体的数据结构，框架在做checkpoint的时候，使用byte...通常在DataStream上的状态推荐使用托管的状态，当实现一个用户自定义的operator时，会使用到原始状态。 顾名思义，就是基于KeyedStream上的状态。...支持不同的重启策略，以在故障发生时控制作业如何重启。 ...默认的重启策略可以通过 Flink 的配置文件 flink-conf.yaml 指定。配置参数 restart-strategy 定义了哪个策略被使用。...在升级的情况下使用 注意：为了能够在作业的不同版本之间以及 Flink 的不同版本之间顺利升级，强烈推荐程序员通过 uid(String) 方法手动的给算子赋予 ID，这些 ID 将用于确定每一个算子的状态范围
领取专属 10元无门槛券
手把手带您无忧上云
