Exactly Once 语义在 Flink 中的实现|青训营笔记

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: 本篇文章主要讲述了Flink是如何实现在分布式环境下,对于task的处理做到exactly-once的语义的(结合二阶段提交协议)

课程资料

课程链接:https://live.juejin.cn/4354/yc_Once

课程PPT:https://bytedance.feishu.cn/file/boxcnFPburXr95rMNel1SHOvISg

学员手册:https://juejin.cn/post/7122754431371706404#heading-36

完整手册:https://bytedance.feishu.cn/docx/doxcnECGEFkCKYqbxaDipK1qrVf

一、数据流和动态表

1.1 在流上定义表

image-20220728150209960

  • 动态表:与表示批处理数据的静态表不同,动态表是随时间变化的。可以像查询静态批处理表一样查询它们
  • 数据库表是 INSERT 、UPDATE 和 DELETE DML 语句的stream的结果,通常称为changelog stream

image-20220728150721680

  • 查询从不终止
  • 查询结果会不断更新,产生一个新的动态表

1.2 不同的连续查询对比

image-20220728150943428

  • 第一个查询更新先前输出的结果,即定义结果表的changelog流包含 INSERT 和 UPDATE 操作
  • 第二个查询只附加到结果表,即结果表的 changelog 流只包含 INSERT 操作

image-20220728151353010

  • Retract Stream:Retract 流(同时包含 INSERT 消息和 DELETE 消息)

二、Exactly-Once 和 Checkpoint

2.1 一致性保证语义

  • At-most-once:出现故障的时候,啥也不做。数据处理不保证任何语义,处理时延低
  • At-least-once:保证每条数据均至少被处理一次,一条数据可能存在重复消费
  • Exactly-once:最严格的处理语义,从输出结果来看,每条数据均被消费且仅消费一次,仿佛故障从未发生

2.2 Checkpoint

  • Checkpoint Barrier 的下发,source 算子接收到 JM 发送的 Checkpoint Barrier 标识状态快照制作的开始

image-20220728152259738

  • 各个 source 保存自己状态后,向所有链接的下游继续发送 Checkpoint Barrier,同时告知 JM 自己状态已经制作完成

image-20220728152550775

  • 算子会等待所有上游的 barrier 到达后才开始快照的制作,已经制作完成的上游算子会继续处理数据,并不会被下游算子制作快照的过程阻塞

image-20220728152837046

  • Checkpoint并不阻塞在算子处

image-20220728153618494

  • 所有算子都告知JM状态制作完成后,整个Checkpoint就都结束了

image-20220728154110156

2.3 Checkpoint 对作业性能的影响

  1. 解耦了快照制作和数据处理过程,各个算子制作完成状态快照后就可以正常处理数据,不用等下游算子制作完成快照
  2. 在快照制作和 Barrier Alignment 过程中需要暂停处理数据,仍然会增加数据处理延迟
  3. 快照保存到远端也有可能极为耗时

三、端到端 Exactly-Once 实现

3.1 Exactly-once语义

image-20220728154333644

  • Checkpoint 能保证每条数据都对各个有状态的算子更新一次,sink输出算子仍然可能下发重复数据
  • 严格意义的端到端的 Exactly-once 语义需要特殊的 sink 算子实现

3.2 两阶段提交协议

在多个节点参与执行的分布式系统中,为了协调每个节点都能同时执行或者回滚某个事务性的操作,引入了一个中心节点来统一处理所有节点的执行逻辑,这个中心节点叫做协作者(coordinater),被中心节点调度的其他业务节点叫做参与者(participater)

image-20220728154913195

  • 预提交阶段

    1. 协作者向所有参与者发送一个 commit 消息
    2. 每个参与者接收到消息后,执行事务,但是不正真提交
    3. 若事务成功执行完成,发送一个成功的消息(vote yes);执行失败,则发送一个失败的消息(vote no)
  • 提交阶段

    1. 协作者向所有参与者发送一个 commit 消息
    2. 每个收到 commit 消息的参与者释放执行事务所需的资源,并结束这次事务的执行
    3. 完成步骤2后,参与者发送一个 ack 消息给协作者
    4. 协作者收到所有参与者的 ack 消息后,标识该事务执行完成
  • 若协作者收到参与者 vote no 的消息(或者发生等待超时)

    1. 协作者向所有参与者发送一个 rollback 消息
    2. 每个收到 rollback 消息的参与者回滚事务的执行操作,并释放事务所占用资源
    3. 完成步骤2之后,参与者发送一个 ack 消息给协作者
    4. 协作者收到所有参与者的 ack 消息之后,标识该事务成功完成回滚

3.3 两阶段提交协议在Flink中的应用

  • Flink 中协作者和参与者的角色分配

image-20220728160048845

  • 协作者(JM)发起第一阶段的提交

image-20220728160344559

  • 各个算子Checkpoint 的制作(类似 vote 的过程,快照制作成功/失败都向 JM 发送消息)

image-20220728160418720

  • 第二阶段提交以及 Checkpoint 制作完成

image-20220728160516683

3.4 Flink 两阶段提交总结

  1. 事务开启:在 sink task 向下游写数据之前,均会开启一个事务,后续所有写数据的操作均在这个事务中执行,事务未提交之前,事务写入的数据下游不可读
  2. 预提交阶段:JM 开始下发 Checkpoint Barrier,当各个处理逻辑接收到 barrier 后,停止处理后续数据,对当前状态制作快照,此时 sink 也不在当前事务下继续处理数据(处理后续的数据需要重新打开下一个事务)。状态制作成功后则向 JM 发送成功消息,失败则发送失败消息
  3. 提交阶段:若 JM 收到所有预提交成功的消息,则向所有处理逻辑(包括 sink )发送可以提交此次事务的消息,sink 接收到此消息后,则完成此次事务的提交,此时下游可以读到这次事务写入的数据;若 JM 有接收到预提交失败的消息,则通知所有处理逻辑回滚这次事务操作,此时 sink 则丢弃这次事务提交的数据
相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
相关文章
|
1月前
|
数据处理 Apache 流计算
Flink Watermark和时间语义
Flink Watermark和时间语义
35 2
|
1月前
|
Apache 流计算
【Flink】Flink的三种时间语义
【4月更文挑战第19天】【Flink】Flink的三种时间语义
|
22天前
|
数据处理 Apache 流计算
Flink Watermark和时间语义
Flink Watermark和时间语义
|
1月前
|
存储 缓存 算法
[尚硅谷flink] 检查点笔记
[尚硅谷flink] 检查点笔记
|
1月前
|
存储 传感器 消息中间件
[尚硅谷 flink] 状态管理 笔记
[尚硅谷 flink] 状态管理 笔记
|
1月前
|
Java 数据处理 分布式数据库
Flink中的Exactly-Once语义是什么?请解释其作用和实现原理。
Flink中的Exactly-Once语义是什么?请解释其作用和实现原理。
44 0
|
1月前
|
存储 算法 Java
读Flink源码谈设计:Exactly Once
将Flink应用至生产已有一段时间,刚上生产的时候有幸排查过因数据倾斜引起的Checkpoint超时问题——当时简单的了解了相关机制,最近正好在读Flink源码,不如趁这个机会搞清楚。 在这里,我们首先要搞清楚两种Exactly-Once的区别: - Exactly Once:在计算引擎内部,数据不丢失不重复。本质是通过Flink开启检查点进行Barrier对齐,即可做到。 - End to End Exactly Once:这意味着从数据读取、引擎处理到写入外部存储的整个过程中,数据都是不丢失不重复的。这要求数据源可重放,写入端支持事务的恢复和回滚或幂等。
540 0
|
1月前
|
消息中间件 Kafka 程序员
Flink(九)【时间语义与水位线】
Flink(九)【时间语义与水位线】
|
1天前
|
消息中间件 Java 关系型数据库
实时计算 Flink版操作报错合集之从 PostgreSQL 读取数据并写入 Kafka 时,遇到 "initial slot snapshot too large" 的错误,该怎么办
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
114 0
|
1天前
|
监控 Oracle 关系型数据库
实时计算 Flink版操作报错合集之在配置连接时,添加了scan.startup.mode参数后,出现报错。是什么导致的
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
123 0