Flink实战(11)-Exactly-Once语义之两阶段提交

0 大纲

[Apache Flink]2017年12月发布的1.4.0版本开始,为流计算引入里程碑特性:TwoPhaseCommitSinkFunction。它提取了两阶段提交协议的通用逻辑,使得通过Flink来构建端到端的Exactly-Once程序成为可能。同时支持:

  • 数据源(source)
  • 和输出端(sink)

包括Apache Kafka 0.11及更高版本。它提供抽象层,用户只需实现少数方法就能实现端到端Exactly-Once语义。

新功能及Flink实现逻辑:

  • 描述Flink checkpoint机制如何保证Flink程序结果的Exactly-Once的
  • 显示Flink如何通过两阶段提交协议与数据源和数据输出端交互,以提供端到端的Exactly-Once保证
  • 通过一个简单的示例,了解如何使用TwoPhaseCommitSinkFunction实现Exactly-Once的文件输出

1 Flink应用中的Exactly-Once语义

Exactly-Once,指每个输入的事件只影响最终结果一次。即使机器或软件故障,既没有重复数据,也不会丢数据。

Flink很久就提供Exactly-Once,checkpoint机制是Flink有能力提供Exactly-Once语义的核心。

一次checkpoint是以下内容的一致性快照:

  • 应用程序的当前状态
  • 输入流的位置

Flink可配置一个固定时间点,定期产生checkpoint,将checkpoint的数据写入持久存储系统,如S3或HDFS。将checkpoint数据写入持久存储是异步,即Flink应用程序在checkpoint过程中可以继续处理数据。

如果发生机器或软件故障,重新启动后,Flink应用程序将从最新的checkpoint点恢复处理; Flink会恢复应用程序状态,将输入流回滚到上次checkpoint保存的位置,然后重新开始运行。这意味着Flink可以像从未发生过故障一样计算结果。

Flink 1.4.0前,Exactly-Once语义仅限Flink应用程序内部,没有扩展到Flink数据处理完后发送的大多数外部系统。Flink应用程序与各种数据输出端进行交互,开发人员自己维护组件上下文保证Exactly-Once语义。

为提供端到端的Exactly-Once语义 – 即除了Flink应用程序内部,Flink写入的外部系统也需要能满足Exactly-Once语义 – 这些外部系统必须提供提交或回滚的方法,然后通过Flink的checkpoint机制协调。

分布式系统中,协调提交和回滚的常用方法是2pc协议。讨论Flink的TwoPhaseCommitSinkFunction如何利用2pc提供端到端的Exactly-Once语义。

2 Flink应用程序端到端的Exactly-Once语义

Kafka经常与Flink使用。Kafka 0.11版本添加事务支持。这意味着现在通过Flink读写Kafaka,并提供端到端的Exactly-Once语义有了必要支持。

Flink对端到端的Exactly-Once语义的支持不仅局限Kafka,可将它与任何一个提供必要的协调机制的源/输出端一起使用。如Pravega,来自DELL/EMC的开源流媒体存储系统,通过Flink的TwoPhaseCommitSinkFunction也能支持端到端的Exactly-Once语义。

image-20231124142310942

示例程序有:

  • 从Kafka读取的数据源(Flink内置的KafkaConsumer)
  • 窗口聚合
  • 将数据写回Kafka的数据输出端(Flink内置的KafkaProducer)

要使数据输出端提供Exactly-Once保证,须将所有数据通过一个事务提交给Kafka。提交捆绑了两个checkpoint之间的所有要写数据。这确保在故障时,能回滚写入的数据。但分布式系统中,通常有多个并发运行的写入任务,所有组件须在提交或回滚时“一致”才能确保一致结果。Flink使用2PC及预提交阶段解决这问题。

pre-commit

checkpoint开始时,即2PC的“预提交”阶段。当checkpoint开始时,Flink的JobManager会将checkpoint barrier(将数据流中的记录分为进入当前checkpoint与进入下一个checkpoint)注入数据流。

brarrier在operator之间传递。对每个operator,它触发operator的状态快照写入state backend。

数据源保存了消费Kafka的偏移量(offset),之后将checkpoint barrier传递给下一operator。

这种方式仅适用于operator具有『内部』状态。

内部状态

指Flink state backend保存和管理的。如第二个operator中window聚合算出来的sum值。当一个进程有它的内部状态时,除了在checkpoint前需将数据变更写入state backend,无需在pre-commit阶段执行其他操作。

Flink负责在checkpoint成功时正确提交这些写入或故障时中止这些写入。

image-20231124150402626

3 Flink应用启动pre-commit阶段

当进程具有『外部』状态,需额外处理。外部状态通常以写入外部系统(如Kafka)的形式出现。此时,为提供Exactly-Once保证,外部系统须【支持事务】,才能和两阶段提交协议集成。

示例数据需写入Kafka,因此数据输出端(Data Sink)有外部状态。此时,在预提交阶段:

  • 除了将其状态写入state backend
  • 数据输出端还必须预先提交其外部事务

当checkpoint barrier在所有operator都传递了一遍,并且触发的checkpoint回调成功完成时,预提交阶段结束。所有触发的状态快照都被视为该checkpoint的一部分。checkpoint是整个应用程序状态的快照,包括预先提交的外部状态。若故障,可回滚到上次成功完成快照的时间点。

下一步是通知所有operator,checkpoint已经成功了。这是2PC的提交阶段,JobManager为应用程序中的每个operator发出checkpoint已完成的回调。

数据源和 widnow operator没有外部状态,因此在提交阶段,这些operator不必执行任何操作。但是,数据输出端(Data Sink)拥有外部状态,此时应该提交外部事务。

总结

  • 一旦所有operator完成预提交,就提交一个commit。
  • 如果至少有一个预提交失败,则所有其他提交都将中止,我们将回滚到上一个成功完成的checkpoint。
  • 在预提交成功之后,提交的commit需要保证最终成功 – operator和外部系统都需要保障这点。如果commit失败(例如,由于间歇性网络问题),整个Flink应用程序将失败,应用程序将根据用户的重启策略重新启动,还会尝试再提交。这个过程至关重要,因为如果commit最终没有成功,将会导致数据丢失。

因此,我们可以确定所有operator都同意checkpoint的最终结果:所有operator都同意数据已提交,或提交被中止并回滚。

4 在Flink中实现两阶段提交Operator

完整的实现两阶段提交协议可能有点复杂,这就是为什么Flink将它的通用逻辑提取到抽象类TwoPhaseCommitSinkFunction中的原因。

接下来基于输出到文件的简单示例,说明如何使用TwoPhaseCommitSinkFunction。用户只需要实现四个函数,就能为数据输出端实现Exactly-Once语义:

  • beginTransaction – 在事务开始前,我们在目标文件系统的临时目录中创建一个临时文件。随后,我们可以在处理数据时将数据写入此文件。
  • preCommit – 在预提交阶段,我们刷新文件到存储,关闭文件,不再重新写入。我们还将为属于下一个checkpoint的任何后续文件写入启动一个新的事务。
  • commit – 在提交阶段,我们将预提交阶段的文件原子地移动到真正的目标目录。需要注意的是,这会增加输出数据可见性的延迟。
  • abort – 在中止阶段,我们删除临时文件。

我们知道,如果发生任何故障,Flink会将应用程序的状态恢复到最新的一次checkpoint点。一种极端的情况是,预提交成功了,但在这次commit的通知到达operator之前发生了故障。在这种情况下,Flink会将operator的状态恢复到已经预提交,但尚未真正提交的状态。

我们需要在预提交阶段保存足够多的信息到checkpoint状态中,以便在重启后能正确的中止或提交事务。在这个例子中,这些信息是临时文件和目标目录的路径。

TwoPhaseCommitSinkFunction已经把这种情况考虑在内了,并且在从checkpoint点恢复状态时,会优先发出一个commit。我们需要以幂等方式实现提交,一般来说,这并不难。在这个示例中,我们可以识别出这样的情况:临时文件不在临时目录中,但已经移动到目标目录了。

在TwoPhaseCommitSinkFunction中,还有一些其他边界情况也会考虑在内,请参考Flink文档了解更多信息。

FAQ

flink sink在如果过来一个checkpoint barrier,会去存储state,这个动作会和普通的write并行吗?还是串行?

在Flink的checkpoint机制中,当一个Checkpoint Barrier过来时,sink会触发对状态的snapshot,这个snapshot动作默认是和普通的write操作并行进行的。

具体来说:

  • Flink的checkpoint机制是通过在datastream中注入Checkpoint Barrier来实现的。

  • 当source接收到Checkpoint Barrier时,会将其传递给下游的transformation和sink。

  • 当sink接收到Checkpoint Barrier时,会启动一个新的线程来执行state snapshot(状态保存)。

  • 这个状态snapshot线程会从状态后端Snapshot State,并存储检查点。

  • 而sink的主线程在接收到Checkpoint Barrier时,会继续处理正常的write。

  • 这样,状态snapshot和正常的write操作就是并行进行的。

但是也可以通过Sink的配置来设置snapshot和write的执行策略,主要有两种模式:

  1. 并行模式(默认):snapshot和write同时进行

  2. 串行模式:snapshot完成后再进行write

综上,Flink sink在默认的并行checkpoint模式下,状态snapshot和普通的write操作是并行执行的。可以通过配置来改变其行为。这样可以根据实际需要进行平衡。

总结

  • Flink的checkpoint机制是支持两阶段提交协议并提供端到端的Exactly-Once语义的基础。
  • 这个方案的优点是: Flink不像其他一些系统那样,通过网络传输存储数据 – 不需要像大多数批处理程序那样将计算的每个阶段写入磁盘。
  • Flink的TwoPhaseCommitSinkFunction提取了两阶段提交协议的通用逻辑,基于此将Flink和支持事务的外部系统结合,构建端到端的Exactly-Once成为可能。
  • 从Flink 1.4.0开始,Pravega和Kafka 0.11 producer都提供了Exactly-Once语义;Kafka在0.11版本首次引入了事务,为在Flink程序中使用Kafka producer提供Exactly-Once语义提供了可能性。
  • Kafaka 0.11 producer的事务是在TwoPhaseCommitSinkFunction基础上实现的,和at-least-once producer相比只增加了非常低的开销。

    本文由博客一文多发平台 OpenWrite 发布!

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

如若内容造成侵权/违法违规/事实不符,请联系编程知识网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

柯桥学历提升培训,考研党要不要关闭朋友圈

关掉朋友圈的好处 在众多上岸人的考研经验贴中,可以看到,“适时关闭朋友圈”成为常见内容。 学长学姐给的建议都是不要通过朋友圈去过度关注别人的光鲜生活,也不必一定要向外展示自己的现状,喜怒哀乐自我消化,保持思想…

浅析linux中的信号

人们往往将信号称为“软件中断”,它提供了异步事件的处理机制,这些事件可以来自系统外部(如用户按下ctrlc产生中断符),也可能来自程序或者内核内部的执行动作(如进程除零操作)。进程收到信号&am…

视频图片提取秘籍:从指定时长中提取想要的视频封面

在如今的内容丰富的互联网世界中,视频已经成为大家获取信息、娱乐和交流的重要方式之一。而视频的封面或图片,作为视频内容的“名片”,往往决定了观众是否愿意点击和进一步了解视频内容。视频图片提取,通常也被称为视频封面提取&a…

Flink-简介与基础

Flink-简介与基础 一、Flink起源二、Flink数据处理模式1.批处理2.流处理3.Flink流批一体处理 三、Flink架构1.Flink集群2.Flink Program3.JobManager4.TaskManager 四、Flink应用程序五、Flink高级特性1.时间流(Time)和窗口(Window&#xff0…

【docker】docker安装与优化

目录 一、安装Docker 1、关闭防火墙 2、安装依赖包 3、设置阿里云镜像源 4、安装Docker-CE社区版并设置为开机自启动 5、查看Docker信息 二、设置镜像加速 1、申请加速地址 2、实现加速操作 三、网络优化 1、如何网络优化 2、具体操作 四、docker-server端配置文件…

无人机巡检如何做到实时识别,从数据到模型全流程解读

在数字化和自动化飞速发展的今天,AI识别算法正在加速进入行业生产系统。 基于巡检数据的智能开发,识别算法突破性进展的核心驱动力在于需求——从全天候巡视的平安城市,到潮汐变化的交通网络,从广阔的水域,到繁忙的街道…

通过视频文件地址截取图像生成图片保存为封面图

安装 RPM Fusion 软件库 FFmpeg并不包含在 CentOS 官方软件库中,需要使用第三方软件库安装。可以使用 RPM Fusion 软件库来获取 FFmpeg。 首先,使用以下命令安装 RPM Fusion 软件库: sudo yum install epel-release -y sudo rpm -Uvh https://download1.rpmfusion.org/fre…

X-RAY POC编写

POC(Proof of Concept) - 利用证明 POC,Proof of Concept,意思是 利用证明。这个短语会在漏洞报告中使用,漏洞报告中的POC则是一段说明或者一个攻击的漏洞介绍,使得读者能够确认这个漏洞是真实存在的。 EXP(Exploit) - 漏洞利用…

对象的内部结构

在HotSpot 虚拟机里,对象在堆内存中的存储布局可以划分为三个部分:对象头( Header )、实例数据(Instance Data )和对齐填充( Padding )。 对象头 Mark Word(标记字段&a…

【Spring集成MyBatis】MyBatis注解开发

文章目录 1. MyBatis的常用注解2. 基于注解的MyBatis增删改查增删改查完整代码加载映射关系测试代码 3. MyBatis的注解实现复杂映射开发一对一操作的实现一对一操作实现的第二种方式一对多操作的实现多对多操作实现 1. MyBatis的常用注解 2. 基于注解的MyBatis增删改查 使用注…

张弛声音变现课,枪战电影高能量、快速节奏

在执行枪战片的声音配音任务时,配音员应该致力于传递出戏剧性的紧张氛围与动作场面的激烈感。枪战场景往往是高能量、快速节奏的,这就要求配音不仅要与视觉动作紧密结合,还要通过声音来增强动作的逼真度和观众的紧迫感。以下是针对枪战电影进…

科研/比赛必备工具及系列笔记集合

科研/比赛必备工具及系列笔记集合 零、前言一、常用工具系列1.1 笔记平台使用感受系列1.2 常用开发平台系列 二、论文系列2.1 检索工具系列2.2 投稿调研系列2.3 常见国际期刊/会议2.4 常见中文核心期刊/会议 三、文献系列3.1 画图工具系列3.2 翻译工具系列3.3 英文纠正系列3.4 …