淘宝网网站开发,今天头条新闻,厦门市建设局网站住房保障2018,做网站套路Flink 详解#xff08;二#xff09;#xff1a;核心篇 Ⅱ 22、刚才提到 State#xff0c;那你简单说一下什么是 State。 在 Flink 中#xff0c;状态 被称作 state#xff0c;是用来保存中间的计算结果或者缓存数据。根据状态是否需要保存中间结果#xff0c;分为 无状… Flink 详解二核心篇 Ⅱ 22、刚才提到 State那你简单说一下什么是 State。 在 Flink 中状态 被称作 state是用来保存中间的计算结果或者缓存数据。根据状态是否需要保存中间结果分为 无状态计算 和 有状态计算。
对于流计算而言事件持续产生如果每次计算相互独立不依赖上下游的事件则相同输入可以得到相同输出是无状态计算。如果计算需要依赖于之前或者后续事件则被称为有状态计算。 有状态计算如 sum 求和数据累加等。 23、Flink 状态包括哪些 1 按照由 用户管理 还是 Flink 管理状态可以分为 原始状态 和 托管状态。
原始状态Raw State由用户自行进行管理。托管状态Managed State由 Flink 自行进行管理的 State。
两者区别
从 状态管理方式 来说Managed State 由 Flink Runtime 管理自动存储自动恢复在内存管理上有优化而 Raw State 需要用户自己管理需要自己序列化Flink 不知道 State 中存入的数据是什么结构只有用户自己知道需要最终序列化为可存储的数据结构。从 状态数据结构 来说Managed State 支持已知的数据结构如 Value、List、Map 等。而 Raw State 只支持字节数组所有状态都要转换为二进制字节数组才可以。从 推荐使用场景 来说Managed State 大多数情况下均可使用而 Raw State 是当 Managed State 不够用时比如需要自定义 Operator 时才会使用 Raw State。在实际生产过程中只推荐使用 Managed State。
2State 按照 是否有 key 划分为 KeyedState 和 OperatorState 两种。
KeyedState 特点
只能用在 KeyedStream 上的算子中状态跟特定的 key 绑定。KeyedStream 流上的每一个 key 对应一个 state 对象。若一个 operator 实例处理多个 key访问相应的多个 state可对应多个 state。KeyedState 保存在 StateBackend 中。通过 RuntimeContext 访问实现 Rich Function 接口。支持多种数据结构ValueState、ListState、ReducingState、AggregatingState、MapState。 OperatorState 特点
可以用于所有算子但整个算子只对应一个 state。并发改变时有多种重新分配的方式可选1均匀分配2合并后每个得到全量。实现 CheckpointedFunction 或者 ListCheckpointed 接口。目前只支持 ListState 数据结构。 这里的 fromElements 会调用 FromElementsFunction 的类其中就使用了类型为 ListState 的 operator state。 24、Flink 广播状态了解吗 Flink 中广播状态叫作 BroadcastState。 在广播状态模式中使用。所谓广播状态模式 就是来自一个流的数据需要被广播到所有下游任务在算子本地存储在处理另一个流的时候依赖于广播的数据。下面以一个示例来说明广播状态模式。 上图这个示例包含两个流一个为 Kafka 模型流该模型是通过机器学习或者深度学习训练得到的模型将该模型通过广播发送给下游所有规则算子规则算子将规则缓存到 Flink 的本地内存中另一个为 Kafka 数据流用来接收测试集该测试集依赖于模型流中的模型通过模型完成测试集的推理任务。
广播状态必须是 MapState 类型广播状态模式需要使用 广播函数 进行处理广播函数提供了处理广播数据流和普通数据流的接口。 25、Flink 状态接口包括哪些 在 Flink 中使用状态包含两种状态接口
状态操作接口使用状态对象本身存储、写入、更新数据。状态访问接口从 StateBackend 获取状态对象本身。
1、状态操作接口
Flink 中的状态操作接口面向两类用户即 应用开发者 和 Flink 框架本身。 所以 Flink 设计了两套接口。
1面向开发者 State 接口
面向开发的 State 接口只提供了对 State 中数据的增删改基本操作接口用户无法访问状态的其他运行时所需要的信息。接口体系如下图 2面向内部 State 接口
内部 State 接口是给 Flink 框架使用提供更多的 State 方法可以根据需要灵活扩展。除了对 State 中数据的访问之外还提供内部运行时信息如 State 中数据的序列化器命名空间namespace、命名空间的序列化器、命名空间合并的接口。内部 State 接口命名方式为 InternalxxxState。
2、状态访问接口
有了状态之后开发者自定义 UDFUserDefineFunction用户自定义函数时应该如何访问状态
状态会被保存在 StateBackend 中但 StateBackend 又包含不同的类型。所以 Flink 中抽象了两个状态访问接口OperatorStateStore 和 KeyedStateStore用户在编写 UDF 时就无须考虑到底是使用哪种 StateBackend 类型接口。
1OperatorStateStore 接口原理 OperatorState 数据以 Map 形式保存在内存中并没有使用 RocksDBStateBackend 和 HeapKeyedStateBackend。
2KeyedStateStore 接口原理 KeyedStateStore 数据使用 RocksDBStateBackend 或者 HeapKeyedStateBackend 来存储KeyedStateStore 中创建、获取状态都交给了具体的 StateBackend 来处理KeyedStateStore 本身更像是一个代理。 26、Flink 状态如何存储 在 Flink 中状态存储 被叫做 StateBackend它具备两种能力
在计算过程中提供访问 State 能力开发者在编写业务逻辑中能够使用 StateBackend 的接口读写数据。能够将 State 持久化到外部存储提供容错能力。
Flink 状态提供三种存储方式
内存型MemoryStateBackend适用于验证、测试、不推荐生产使用。文件型FSStateBackend适用于长周期大规模的数据。RocksDB: RocksDBStateBackend适用于长周期大规模的数据。
上面提到的 StateBackend 是 面向用户 的在 Flink 内部 3 种 State 的关系如下图 在运行时MemoryStateBackend 和 FSStateBackend 本地的 State 都保存在 TaskManager 的内存中所以其底层都依赖于 HeapKeyedStateBackend。HeapKeyedStateBackend 面向 Flink 引擎内部使用者无须感知。
1、内存型 StateBackend
MemoryStateBackend运行时所需的 State 数据全部保存在 TaskManager JVM 堆上内存中KV 类型的 State、窗口算子的 State 使用 HashTable 来保存数据、触发器等。执行检查点的时候会把 State 的快照数据保存到 JobManager 进程的内存中。
MemoryStateBackend 可以使用异步的方式进行快照也可以同步推荐异步避免阻塞算子处理数据。
基于内存的 StateBackend 在生产环境下不建议使用可以在本地开发调试测试 。注意点如下
State 存储在 JobManager 的内存中受限于 JobManager 的内存大小。每个 State 默认 5 M B 5MB 5MB可通过 MemoryStateBackend 构造函数调整。每个 Stale 不能超过 Akka Frame 大小。
2、文件型 StateBackend
FSStateBackend运行时所需的 State 数据全部保存在 TaskManager 的内存中 执行检查点的时候会把 State 的快照数据保存到配置的文件系统中。
可以是分布式或者本地文件系统路径如
HDFS路径“hdfs://namenode:40010/flink/checkpoints”本地路径“file:///data/flink/checkpoints”
FSStateBackend 适用于处理大状态、长窗口、或者大键值状态的有状态处理任务。注意点如下
State 数据首先被存在 TaskManager 的内存中。State 大小不能超过 TM 内存。TM 异步将 State 数据写入外部存储。
MemoryStateBackend 和 FSStateBackend 都依赖于 HeapKeyedStateBackendHeapKeyedStateBackend 使用 State 存储数据。
3、RocksDBStateBackend
RocksDBStateBackend 跟内存型和文件型都不同 。
RocksDBStateBackend 使用嵌入式的本地数据库 RocksDB 将流计算数据状态存储在本地磁盘中不会受限于 TaskManager 的内存大小在执行检查点的时候再将整个 RocksDB 中保存的 State 数据全量或者增量持久化到配置的文件系统中在 JobManager 内存中会存储少量的检查点元数据。RocksDB 克服了 State 受内存限制的问题同时又能够持久化到远端文件系统中比较适合在生产中使用。
缺点RocksDBStateBackend 相比基于内存的 StateBackend访问 State 的成本高很多可能导致数据流的吞吐量剧烈下降甚至可能降低为原来的 1 / 10 1/10 1/10。
适用场景
最适合用于处理大状态、长窗口或大键值状态的有状态处理任务。RocksDBStateBackend 非常适合用于高可用方案。RocksDBStateBackend 是目前唯一支持增量检查点的后端。增量检查点非常适用于超大状态的场景。
注意点
总 State 大小仅限于磁盘大小不受内存限制。RocksDBStateBackend 也需要配置外部文件系统集中保存 State。RocksDB 的 JNI API 基于 byte 数组单 Key 和单 Value 的大小不能超过 8 8 8 字节。对于使用具有合并操作状态的应用程序如 ListState 随着时间可能会累积到超过 2 31 2^{31} 231 字节大小这将会导致在接下来的查询中失败。 27、Flink 状态如何持久化 首先Flink 的状态最终都要持久化到第三方存储中确保集群故障或者作业挂掉后能够恢复。RocksDBStateBackend 持久化策略有两种
全量持久化策略RocksFullSnapshotStrategy增量持久化策略RocksIncementalSnapshotStrategy
1、全量持久化策略
每次将全量的 State 写入到状态存储中HDFS。内存型、文件型、RocksDB 类型的 StataBackend 都支持全量持久化策略。 在执行持久化策略的时候使用异步机制每个算子启动 1 1 1 个独立的线程将自身的状态写入分布式存储可靠存储中。在做持久化的过程中状态可能会被持续修改基于内存的状态后端使用 CopyOnWriteStateTable 来保证线程安全RocksDBStateBackend 则使用 RocksDB 的快照机制使用快照来保证线程安全。
2、增量持久化策略
增量持久化就是每次持久化增量的 State只有 RocksDBStateBackend 支持增量持久化。
Flink 增量式的检查点以 RocksDB 为基础 RocksDB 是一个基于 LSM-Tree 的 KV 存储。新的数据保存在内存中 称为 memtable。如果 Key 相同后到的数据将覆盖之前的数据一旦 memtable 写满了RocksDB 就会将数据压缩并写入磁盘。memtable 的数据持久化到磁盘后就变成了不可变的 sstable。
因为 sstable 是不可变的Flink 对比前一个检查点创建和删除的 RocksDB sstable 文件就可以计算出状态有哪些发生改变。
为了确保 sstable 是不可变的Flink 会在 RocksDB 触发刷新操作强制将 memtable 刷新到磁盘上。在 Flink 执行检查点时会将新的 sstable 持久化到 HDFS 中同时保留引用。这个过程中 Flink 并不会持久化本地所有的 sstable因为本地的一部分历史 sstable 在之前的检查点中已经持久化到存储中了只需增加对 sstable 文件的引用次数就可以。
RocksDB 会在后台合并 sstable 并删除其中重复的数据。然后在 RocksDB 删除原来的 sstable替换成新合成的 sstable。新的 sstable 包含了被删除的 sstable 中的信息通过合并历史的 sstable 会合并成一个新的 sstable并删除这些历史 sstable。可以减少检查点的历史文件避免大量小文件的产生。 28、Flink 状态过期后如何清理 1、DataStream 中状态过期
可以对 DataStream 中的每一个状态设置清理策略 StateTtlConfig可以设置的内容如下
过期时间超过多长时间未访问视为 State 过期类似于缓存。过期时间更新策略创建和写时更新、读取和写时更新。State 可见性未清理可用超时则不可用。
2、Flink SQL 中状态过期
Flink SQL 一般在流 Join、聚合类场景使用 State如果 State 不定时清理则导致 State 过多内存溢出。清理策略配置如下
StreamQueryConfig qConfig ...
//设置过期时间为 min 12小时 max 24小时
qConfig.withIdleStateRetentionTime(Time.hours(12),Time.hours(24));
