Flink 状态管理与容错机制(CheckPoint & SavePoint)的关系

这篇具有很好参考价值的文章主要介绍了Flink 状态管理与容错机制(CheckPoint & SavePoint)的关系。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

一、什么是状态

无状态计算的例子: 例如一个加法算子,第一次输入2+3=5那么以后我多次数据2+3的时候得到的结果都是5。得出的结论就是,相同的输入都会得到相同的结果,与次数无关。
有状态计算的例子: 访问量的统计,我们都知道Nginx的访问日志一个请求一条日志,基于此我们就可以统计访问量。如下,/api/a这个url第一此访问的时候,返回的结果就是 count1,但当第二次访问的时候,返回的结果变成了2。为什么Flink知道之前已经处理过一次 hello world,这就是state发挥作用了,这里是被称为keyed state存储了之前需要统计的数据,keyby接口的调用会创建keyed streamkey进行划分,这是使用keyed state的前提。得出的结论就是,相同的输入得到不同的结果,与次数有关。这就是有状态的数据。
Flink 状态管理与容错机制(CheckPoint & SavePoint)的关系,Flink,flink,大数据,后端,java,面试,性能优化,hadoop

什么场景下会大量使用到这种状态数据啦?简单举几个例子:
【1】去重的需求中,比如说我们只想知道这100个同事都属于那几个部门的等等。
【2】窗口计算,已进入未触发的数据。比如,我们一分钟统计一次,1-2之间的1.5这个时候的数据对于2来说就是一个有状态的数据,因为2的结果与1.5有关。
【3】机器学习/深度学习,训练的模型及参数。这对于机器学习的同学深入感触。比如,第一次输入hello,机器会给我一个反馈,那么下次会基于这个反馈做进一步的学习处理。那么上一步的结果对于我而言就是一种有状态的输入。
【4】访问历史数据,需要与昨日进行对比。昨日的数据对于今日而言也属于一种状态。你品,你细品。

为什么要管理状态,用内存不香吗?首先流失作业是有它的标准的,不是什么东西随随便便就说自己这个是流失处理。首先,7*24小时运行,高可靠,你内存不行吧,你的容量总有用完的时候吧。其次,数据不丢失不重,恰好计算一次,你内存要实现需要备份和恢复,你还总伴随着小部分数据的丢失吧。最后,数据实时产生,不延迟,你内存不够横向扩展时,你需要延迟吧。

理想的状态管理就是下面描述的样子,Flink也都帮我们实现了。
Flink 状态管理与容错机制(CheckPoint & SavePoint)的关系,Flink,flink,大数据,后端,java,面试,性能优化,hadoop

二、状态的类型

Managed State & Raw State

Managed State Raw State
状态管理方式 Flink Runtime 管理 —自动存储,自动恢复 —内存管理上有优化 用户自己管理(Flink不知道你在State中存储的数据结构的) —要自己实例化
状态数据结构 已知的数据结构 —value,list,map… 字节数据 —byte[]
推荐使用场景 大多数情况下均可使用 自定义 Operator 时可以使用(当Managed State 不够时使用)

Managed Stated 分为: Keyed StatedOperator State
【1】Keyed Stated: 只能用于keyBy生成的KeyedStream上的算子。每一个key对应一个State,一个Operator实例处理多个Key,访问相应的多个State。相同Key会在相同的实例中处理。整个过程如果没有keyBy操作,它是没有KeyedStream的,而Keyed Stated只能应用在KeyedStream 上。

并发改变: State随着Key在实例间迁移。例如:实例A中之前处理KeyAKeyB,后面我扩展了实例B,那么 实例A就只需要处理KeyAKeyB就交给 实例B进行处理。安装状态进行分离,可以理解为分布式。

通过 RuntimeContext 访问,说明Operator是一个Rich Function,否则是拿不到RuntimeContext

支持的数据结构: ValueStateListStateReducingStateAggregatingStateMapState

【2】Operator State: 可以用于所有的算子,常用于source上,例如FlinkKafkaConsumer。一个Operator实例对应一个State,所以一个Operator中会处理多个key,可以理解为集群。

并发改变: Operator State没有key,并发改变的时候就需要重新分配。内置了两种方案:均匀分配和合并后每个得到全量。

访问方式: 实现CheckpointedFunctionListCheckpointed接口。

支持的数据结构: ListState

三、Keyed State 使用示例

什么是 keyed state: 对于keyed state,有两个特点:
【1】只能应用于KeyedStream 的函数与操作中,例如Keyed UDF, window state
【2】keyed state是已经分区 / 划分好的,每一个 key 只能属于某一个 keyed state
对于如何理解已经分区的概念,我们需要看一下keyby的语义,大家可以看到下图左边有三个并发,右边也是三个并发,左边的词进来之后,通过keyby会进行相应的分发。例如对于hello wordhello这个词通过hash运算永远只会到右下方并发的task上面去。
Flink 状态管理与容错机制(CheckPoint & SavePoint)的关系,Flink,flink,大数据,后端,java,面试,性能优化,hadoop

什么是 operator state
【1】又称为non-keyed state,每一个operator state都仅与一个operator的实例绑定。
【2】常见的operator statesource state,例如记录当前sourceoffset再看一段使用operator stateword count代码:
Flink 状态管理与容错机制(CheckPoint & SavePoint)的关系,Flink,flink,大数据,后端,java,面试,性能优化,hadoop

这里的fromElements会调用FromElementsFunction的类,其中就使用了类型为list stateoperator state。如下几种Keyed State之间的依赖关系,都是state的子类。它们的访问方式和数据结构都有一定的区别。
Flink 状态管理与容错机制(CheckPoint & SavePoint)的关系,Flink,flink,大数据,后端,java,面试,性能优化,hadoop

状态数据类型 访问接口 备注
ValueState 单个值 [update(T) 修改/T value 获取] 例如 WordCount 用 word 做 key,state就是单个的数值。这个单个也可以是字符串、对象等都有可能。访问方式只有上面两种。
MapState Map put(UK key, UV value) putAll(Map<UK,UV> map) remove(UK key) boolean contains(UK key) UV get(UK key) Iterable<Map.Entry> entries() Iterable<Map.Entry> iterator() Iterable keys() Iterable values() 能够操作具体的对象的key
ListState List add/ addAll(List) update(List) Iterable get()
ReducingState 单个值 add/ addAll(List) update(List) T get() 与 List 是同一个父类,这个add是直接将数据更新进了 Reducing的结果里面。举个例子,例如我们统计1分钟的结果,list是先将数据添加到list中,等到1分钟的时候全来出来统计。而 Reducing是来一条就统计一条结果。好处是节省内存。
AggregatingState 单个值 add(IN)/OUT get() 与 List 是同一个父类,与Reducing的不同是,Reducing输入和输出的类型都是相同的。而Aggregating 是可以不同的。例如,我要计算一个平局值,Reducing是算好返回,而Aggregating会返回总和和个数。

举个ValueState的案例

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//获取数据流
DataStream<Event> events = env.addSource(source);

DataStream<Alert> alerts = events
        // 生成 keyedStata 通过 sourceAddress
        .keyBy(Event::sourceAddress)
        // StateMachineMapper 状态机
        .flatMap(new StateMachineMapper());


//我么看下状态机怎么写   实现 RichFlatMapFunction
@SuppressWarnings("serial")
static class StateMachineMapper extends RichFlatMapFunction<Event, Alert> {

    private ValueState<LeaderLatch.State> currentState;

    @Override
    public void open(Configuration conf) {
        // 获取一个 valueState
        currentState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("state", State.class));
    }

    //来一条数据处理一条
    @Override
    public void flatMap(Event evt, Collector<Alert> out) throws Exception {
        // 获取 value
        State state = currentState.value();
        if (state == null) {
            state = State.Initial;//State 是本地的变量
        }

        // 把事件对状态的影响加上去,得到一个状态
        State nextState = state.transition(evt.type());

        //判断状态是否合法
        if (nextState == State.InvalidTransition) {
            //扔出去
            out.collect(new Alert(evt.sourceAddress(), state, evt.type()));
        }
        //是否不能继续转化了,例如取消的订单
        else if (nextState.isTerminal()) {
            // 从 state 中清楚掉
            currentState.clear();
        }
        else {
            // 修改状态
            currentState.update(nextState);
        }
    }
}

四、CheckPoint 与 state 的关系

Checkpoint是从source触发到下游所有节点完成的一次全局操作。下图可以有一个对Checkpoint的直观感受,红框里面可以看到一共触发了 569KCheckpoint,然后全部都成功完成,没有fail的。
Flink 状态管理与容错机制(CheckPoint & SavePoint)的关系,Flink,flink,大数据,后端,java,面试,性能优化,hadoop

**state 其实就是 Checkpoint 所做的主要持久化备份的主要数据,**看下图的具体数据统计,其state也就9kb大小 。
Flink 状态管理与容错机制(CheckPoint & SavePoint)的关系,Flink,flink,大数据,后端,java,面试,性能优化,hadoop

五、状态如何保存和恢复

Checkpoint定时制作分布式快照,对程序的状态进行备份。发生故障时,将整个作业的Task都回滚到最后一次成功Checkpoint中的状态,然后从保存的点继续处理。

必要条件: 数据源支持重发(如果不重发,丢失的消息就真的丢了)

一致性语义: 恰好一次(如果p相同,单线程,多个线程时,可能有的算子对其已经计算了一次了,有的没有就需要注意),至少一次。

//  获取运行环境
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//状态数据
//两个checkpoint 触发间隔设置1S,越频繁追的数据就越少,io消耗也越大
env.enableCheckpointing(1000);
//EXACTLY_ONCE语义说明 Checkpoint是要对替的,这样消息不会重复,也不会对丢。
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//两个checkpoint 最少等待500ms 例如第一个checkpoint做了700ms按理300ms后就要做下一个checkpoint。但是它们之间的等待时间300ms<500ms 此时,就会延长200ms减少checkpoint过于频繁,影响业务。
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
//checkpoint多久超时,如果这个checkpoint在1分钟内还没做完,那就失败了
env.getCheckpointConfig().setCheckpointTimeout(60000);
//同时最多有多少个checkpoint进行
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
//当重新分配并发度,拆分task时,是否保存checkpoint。如果不保存就需要使用savepoint来保存数据,放到外部的介质中。
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);

Checkpoint vs Savepoint

Checkpoint Savepoint
触发管理方式 由Flink自动触发并管理 由用户手动触发并管理
主要用途 在 Task 发生异常时快速恢复,例如网络抖动导致的超时异常 有计划的进行备份,使作业能停止后再恢复,例如修改代码、调整并发。
特点 轻量、自动从故障中服务、在作业停止后默认清除 持久、以标准格式存储,允许代码或配置发生变化、手动触发 savepoint 恢复。

可选的状态存储方式:
【1】MemoryStateBackend:构造方法:

MemoryStateBackend(int maxStateSize, boolean asynchronousSnapshots)

存储方式: StateTaskManager内存。CheckpointJobManager内存。
容量限制: 单个State maxStateSize默认5MmaxStateSize <= akka.framesize默认10M。总大小不超过JobManager内存。
推荐使用场景: 本地测试,几乎无状态的作业,比如ETL/JobManager不容易挂,或影响不大的情况。不推荐在生产场景使用。

【2】FsStateBackend: 构造方法:

FsStateBackend(URL checkpointDataUri, boolean asynchronousSnapshots)

存储方式: StateTaskManager内存。Checkpoint:外部文件系统(本地或HDFS)。
容量限制: 单个TaskManagerState总量不超过它的内存。总大小不超过配置的文件系统容量(会定期清理)。
推荐使用场景: 常规使用状态的作业,例如分钟级窗口聚合、join。需要开启HA的作业。可以在生产环境使用。

【3】RocksDBStateBackend: 构造方法:

RocksDBStateBackend(URL checkpointDataUri, boolean enableIncrementalCheckpointing)

存储方式: StateTaskManager上的KV数据库(实际使用内存+磁盘)。Checkpoint:外部文件系统(本地或HDFS)。
容量限制: 单个TaskManagerState总量不超过它的内存+磁盘,单个key 最大2G。总大小不超过配置的文件系统容量。
推荐使用场景: 超大状态的作业,例如天级窗口聚合。需要开启HA的作业。对状态读写性能要求比较高的作业。可以在生产环境使用。文章来源地址https://www.toymoban.com/news/detail-760551.html

到了这里,关于Flink 状态管理与容错机制(CheckPoint & SavePoint)的关系的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 209.Flink(四):状态,按键分区,算子状态,状态后端。容错机制,检查点,保存点。状态一致性。flink与kafka整合

    算子任务可以分为有状态、无状态两种。 无状态:filter,map这种,每次都是独立事件 有状态:sum这种,每次处理数据需要额外一个状态值来辅助。这个额外的值就叫“状态” (1)托管状态(Managed State)和原始状态(Raw State) 托管状态 就是由Flink统一管理的,状态的存储访问

    2024年02月06日
    浏览(50)
  • 【Flink状态管理五】Checkpoint的设计与实现

    由于系统原因导致Flink作业无法正常运行的情况非常多,且很多时候都是无法避免的。对于Flink集群来讲,能够快速从异常状态中恢复,同时保证处理数据的正确性和一致性非常重要。Flink主要借助Checkpoint的方式保障整个系统状态数据的一致性,也就是基于ABS算法实现轻量级快

    2024年02月21日
    浏览(38)
  • Flink---13、容错机制(检查点(保存、恢复、算法、配置)、状态一致性、端到端精确一次)

                           星光下的赶路人star的个人主页                        大鹏一日同风起,扶摇直上九万里 在Flink中,有一套完整的容错机制来保证故障后的恢复,其中最重要的就是检查点。 1.1.1 检查点的保存 1、周

    2024年02月08日
    浏览(49)
  • 【Flink状态管理(六)】Checkpoint的触发方式(1)通过CheckpointCoordinator触发算子的Checkpoint操作

    Checkpoint的触发方式有两种 一种是数据源节点中的Checkpoint操作触发,通过CheckpointCoordinator组件进行协调和控制。 CheckpointCoordinator通过注册定时器的方式按照 配置的时间间隔触发数据源节点的Checkpoint操作 。数据源节点会向下游算子发出Checkpoint Barrier事件,供下游节点使用。

    2024年02月21日
    浏览(41)
  • Flink状态管理与检查点机制

    本专栏案例代码和数据集链接:  https://download.csdn.net/download/shangjg03/88477960 相对于其他流计算框架,Flink 一个比较重要的特性就是其支持有状态计算。即你可以将中间的计算结果进行保存,并提供给后续的计算使用: 具体而言,Flink 又将状态 (State) 分为 Keyed State 与 O

    2024年02月07日
    浏览(43)
  • Flink理论—容错之状态

    在 Flink 的框架中,进行有状态的计算是 Flink 最重要的特性之一。所谓的 状态,其实指的是 Flink 程序的中间计算结果 。Flink 支持了不同类型的状态,并且针对状态的持久化还提供了专门的机制和状态管理器。 Flink 使用流重放 和 检查点 的组合来实现容错。检查点标记每个输

    2024年02月20日
    浏览(39)
  • Flink-容错机制

    流式数据连续不断地到来,无休无止;所以流处理程序也是持续运行的,并没有一个明确的结束退出时间。机器运行程序,996 起来当然比人要容易得多,不过希望“永远运行”也是不切实际的。因为各种硬件软件的原因,运行一段时间后程序可能异常退出、机器可能宕机,

    2024年02月02日
    浏览(43)
  • Flink中的容错机制

    在Flink中,有一套完整的容错机制来保证故障后的恢复,其中最重要的就是检查点。 在流处理中,我们可以用存档读档的思路,将之前某个时间点的所有状态保存下来,这份存档就被称为“检查点(CkeckPoint)”。 当Flink程序异常重启时,我们就可以在检查点中“ 读档 ”,恢

    2024年01月23日
    浏览(42)
  • 【Flink】容错机制

    目录 1、检查点 ​编辑1.1 检查点的保存 1.1.1 周期性的触发保存 1.1.2 保存的时间点 1.1.3 时间点的保存与恢复 1.1.3.1保存 ​编辑 1.1.3.2 恢复的具体步骤: 1.2 检查点算法 1.2.1 检查点分界线(Barrier) 1.2.2 分布式快照算法(Barrier对齐的精准一次) 1.2.3 分布式快照算法(Barrier对

    2024年02月04日
    浏览(38)
  • Flink容错机制

    在出现故障时,我们将系统重置回正确状态,以确保数据的完整性和准确性。在流处理中,我们采用存档和读档的策略,将之前的计算结果进行保存。这样,在系统重启后,我们可以继续处理新数据,而无需重新计算。 更重要的是,在有状态的流处理中,任务需要保持其之前

    2024年01月20日
    浏览(39)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包