Flink状态(二)

news/2024/9/29 3:33:49

Flink提供了不同的状态存储方式,并说明了状态如何存和存储在哪里。
状态可以被存储在Jvm的堆和堆外。根据状态存储方式的不同,Flink也能代替应用管理状态,意思是Flink能够进行内存管理(有必要的时候,可能会溢出到硬盘),允许应用保存非常大的状态。默认情况下,在配置文件flink-conf.yaml中为所有Flink作业配置状态存储方式。

然而,默认的状态存储方式配置可以被单独的作业设置覆盖,就像下面那样。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(...);

使用Data Stream API写的程序经常需要以多种情况保存状态:

  • 在窗口被触发之前, 窗口需要保存或聚合元素
  • 转换算子也许会使用key/value状态接口保存数据
  • 转换算子也许实现CheckpointedFunction接口使本地变量容错。

当checkpointing被激活的时候,一旦发生checkpoint,状态会被保存,这样数据就不会丢失,并且在恢复的时候能够保持数据一致性。状态在内部是怎么表示的,以及当checkpoint的时候,状态怎么样被保存,以及保存到哪里依赖选择的状态存储方式。

Flink提供了三种开箱即用的状态存储方式:

  • MemoryStateBackend 内存存储
  • FsStateBackend 文件系统存储
  • RocksDBStateBackend RocksDB存储

如果没有特殊配置,系统默认使用内存存储方式。

MemoryStateBackend 内存存储

内存存储:在Java堆中保存状态对象。Key/Value状态和窗口算子都会以Hash表的方式保存状态值,触发器等。
当checkpoint的时候,状态存储将会快照状态,将当checkpoint向JobManager发送回执消息时,作为消息的一部分发给JobManager(master),JobManager会将状态存储到堆内存中。

可以配置内存存储使用异步快照。我们也强烈推荐使用异步快照,避免阻塞流处理通道。请注意默认是打开异步快照的。如果想要关闭这个特性,用户可以在实现化MemoryStateBackend对象的时候,给构造函数中相应的boolean参数传false(这应该仅用于调试目的)。

new MemoryStateBackend(MAX_MEM_STATE_SIZE, false);

内存存储有如下限制:

  • 每一个状态大小默认不超过5M。这个值可以在实例化MemoryStateBackend的时候增加
  • 不管配置的最大状态大小是多少,状态大小不能超过akka配置的桢(一次RPC传输的数据)大小(参数: akka.framesize,默认:10M)。
  • 聚合的状态必须适合JobMaanger内存

以下情况推荐使用内存存储

  • 本地开发或调用
  • 只保存少量状态的作业。例如仅仅包含一次一条记录算子(例如:Map,FlatMap,Fliter,....)的作业。对于这样的作业,Kafka Consumer 仅仅需要非常少的状态。

FsStateBackend 文件系统存储

通过配置文件系统的URL(类型,地址,路径)使用文件系统存储。例如"hdfs://namenode:40010/flink/checkpoints"或者"file:///data/flink/checkpoints"
FsStateBackend 将状态数据保存在TaskManager’s 内存中。当checkpoint的时候,将状态数据写到配置的文件系统或目录中。最小的元数据会存储到JobManager内存中(或者在HA模式下,存储到checkpoint元数据中).
FsStateBackend 默认使用异步快照,以避免阻塞流处理。如果想禁止该特性,在实现化FsStateBackend对象的时候,构造函数中应的参数传入false即可。

new FsStateBackend(path, false);

以下情况,推荐使用FsStateBackend

  • 具有大状态,长窗口,大的key/value状态的作业
  • 所有HA模式下

RocksDBStateBackend RocksDB存储

要想使用RocksDB存储,需要配置文件系统的URL(类型,地址,路径)。例如"hdfs://namenode:40010/flink/checkpoints"或者"file:///data/flink/checkpoints"
RocksDBStateBackend 将状态数据保存到RocksDB数据库.RocksDB文件默认会存储到TaskManager的数据目录中。当checkpoint的时候,整个RocksDB数据库将会保存到配置的文件系统或目录中。最小的元数据会存储到JobManager内存中(或者在HA模式下,存储到checkpoint元数据中).

RocksDBStateBackend 总是执行异步快照。

RocksDBStateBackend 具有如下限制:

  • 由于 RocksDB JNI通信使用的API基于byte[],每个key或每个value最大支持2^31字节。
    注意: 在以RocksDB作用存储情况下,使用merge操作的状态(例如:ListState)会默默地将值大小累加到大于2^31字节,当再次读取的时候会失败,这是目前RocksDB JNI的限制。

以下情况,推荐使用RocksDBStateBackend

  • 具有非常大的状态,长窗口,大的key/value状态的作业
  • 所有HA模式下

你可以保存的状态数据量仅仅受限于磁盘剩余空间大小。与将状态保存到内存中的``FsStateBackend `相比,可以保存更大的状态。然而这也意味着能达到的最大吞吐量更小。因为所有从rocksDB读或写入rocksDB都需要经过序列化与反序例化,比那些基于Java堆的存储后端开销更大。

RocksDBStateBackend 是目前唯一提供 增量的checkpoint的存储。

RocksDB的一些指标可以被获取,但是默认没打开,可以在这里找到全部文档说明。

配置状态存储

如果你什么也没配置,默认的状态存储在JobManager内存中。如果你希望为所有作业默认一个其它的存储,你可以在flink-conf.ymal中配置其它的存储。当然,每一个作业也能单独设置存储。

每个作业单独设置存储

下面示例显示StreamExecutionEnvironment 的作业如何设置存储。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new FsStateBackend("hdfs://namenode:40010/flink/checkpoints"));

如果你想使用 RocksDBStateBackend ,你就必须在你的Flink项目中添加如下Maven依赖。

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-statebackend-rocksdb_2.11</artifactId><version>1.8.0</version>
</dependency>

设置默认的状态存储

默认的状态存储能够在flink-conf.yaml文件中配置,参数是state.backend. 值可以选择jobmanager(MemoryStateBackend), filesystem(FsStateBackend),rocksdb(RocksDBStateBackend)三者中的一个,也可以配置实现了接口StateBackendFactory的全类名。例如: RocksDBStateBackend 的实现类 org.apache.flink.contrib.streaming.state.RocksDBStateBackendFactory.

state.checkpoints.dir参数定义了checkpoint数据和元数据文件存储的位置,你可以在这里发现更详细的checkpoint目录结构说明

配置示例:

# 状态存储
state.backend: filesystem# checkpoints数据存储目录
state.checkpoints.dir: hdfs://namenode:40010/flink/checkpoints

翻译自: https://ci.apache.org/projects/flink/flink-docs-release-1.8/ops/state/state_backends.html

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.hjln.cn/news/47518.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈,一经查实,立即删除!

相关文章

服务器数据恢复成功案例

一、服务器数据恢复描述 需要进行数据恢复的服务器为OceanStor S6800T,服务器硬件配置情况为15块磁盘组成一组80T大小的raid5阵列,阵列划分为两个lun、三个分区。服务器重装系统后发现磁盘分区丢失了一个,需要进行服务器数据恢复。二、服务器数据备份 服务器数据恢复第一步需…

使用 TensorRT C++ API 调用GPU加速部署 YOLOv10 实现 500FPS 推理速度——快到飞起!!

NVIDIA TensorRT ™ 是一款用于高性能深度学习推理的 SDK,包含深度学习推理优化器和运行时,可为推理应用程序提供低延迟和高吞吐量。YOLOv10是清华大学研究人员近期提出的一种实时目标检测方法,通过消除NMS、优化模型架构和引入创新模块等策略,在保持高精度的同时显著降低…

学习笔记483—如何关闭 Mac Office 更新提示

Mac Office 总是弹出Microsoft AutoUpdate更新提示怎么办? 问题:Mac电脑安装 Office 2021 Mac版后,老是弹出更新提示(Microsoft AutoUpdate),关掉之后再开机又会弹出来,怎么办? 解决方法:用 App Cleaner 把Mac Office 更新程序 Microsoft AutoUpdate 直接卸载,这样就…

Head

知足常乐!

oracle partition by 语法

参考链接:https://www.cnblogs.com/xfeiyun/p/17765225.html 数据准备 建表 create table xzq_person ( --行政区人口表province varchar(10),city varchar(10),persons number(10) );插入数据 insert into xzq_person values (江苏,苏州,10); insert into xzq_person v…

机器学习day04

从机器学习过渡到深度学习,虽然都是初步的就是了……study

服务器存储误删所有lun的数据恢复解决方案

NetApp FAS3220是NetApp推出的中端存储阵列,针对NAS和SAN环境构建,被定制为虚拟化、私有云或传统、早先的用途,适用于从几TB到超过2PB的存储需求,提供数据保护,可扩展性,自动精简配置,精简克隆,备份和灾难恢复,下面就讲解NetApp FAS 3220存储的数据恢复的方法。 本次讲…

前 OpenAI 首席科学家建「安全超级智能」实验室;Meta 重组元宇宙团队丨 RTE 开发者日报 Vol.228

开发者朋友们大家好:这里是 「RTE 开发者日报」 ,每天和大家一起看新闻、聊八卦。我们的社区编辑团队会整理分享 RTE(Real-Time Engagement) 领域内「有话题的新闻」、「有态度的观点」、「有意思的数据」、「有思考的文章」、「有看点的会议」,但内容仅代表编辑的个人观点…