首页 > 其他分享 >Exactly Once 语义在Flink中的实现

Exactly Once 语义在Flink中的实现

时间:2023-10-15 17:07:50浏览次数:44  
标签:事务 Flink 快照 处理 制作 Exactly 算子 数据 Once

数据流和动态表

SQL和流处理的区别

流式数据是一种实时生成的数据,而在一般的数据表中存储的数据肯定是有限的,这就会产生矛盾,由此就需要一种新表来存储流式数据,动态表就产生了。

动态表

动态表与表示批处理数据的静态表不同,动态表是随时间变化的。可以像查询静态批处理表一样查询它们。 连续查询的特点

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

在任何时候,连续查询的结果在语义上与以批处理模式在输入表快照上执行的相同查询的结果相同。

不同数据处理保证的语义

  1. At-most-once:至多一次,表示一条消息不管后续处理成功与否只会被消费处理一次。出现故障的时候,啥也不做。数据处理不保证任何语义,处理时延低;
  2. At-leastonce:保证每条数据均至少被处理一次,一条数据可能存在重复消费。
  3. Exactly-once:最严格的处理语义,从输出结果来看,每条数据均被消费且仅消费一次,仿佛故障从未发生。

Exactly-once和checkpoint

Checkpoint是Flink实现容错机制最核心的功能,它会根据用户的配置周期性地对流中各个算子(Operator)的状态生成快照,持久化到外部存储。Flink程序一旦意外崩溃时,重新运行程序时可以有选择地从这些快照进行恢复,从而修正因为故障带来的程序数据异常。Flink写入到外部存储是异步的,意味着Flink在这个阶段可以继续处理数据。

制作快照的时间点

状态恢复的时间点:需要等待所有处理逻辑消费完成source保留状态及之前的数据。 一个简单的快照制作算法

  1. 暂停处理输入的数据;
  2. 等待后续所有处理算子消费当前已经输入的数据;
  3. 待2处理完后,作业所有算子复制自己的状态并保存到远端可靠存储;
  4. 恢复对输入数据的处理

快照制作的开始

每一个source算子都接收到JM发送的Checkpoint Barrier表示状态快照制作的开始

Source算子的处理

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

Barrier Alignment

算子会等待所有上游的barrier到达后才开始快照的制作;

已经制作完成的上有算子会继续处理数据,并不会被下游算子制作快照的过程阻塞。

快照制作和处理数据的解耦

checkpoint的结束

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

CheckPoint对作业性能的影响

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

Flink端到端的Exactly-once语义

端到端Exactly-once语义

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

两阶段提交协议

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

预提交阶段

  • 协作者向所有参与者发送一个commit消息;
  • 每个参与的写作者收到消息后,执行事务,但是不真正提交;
  • 发送事务执行的状态信息

提交阶段

当写作和成功接收到所有的参与者 vote yes的消息:

  • 协作者向所有参与者发送一个commit消息;
  • 每个收到commit消息的参与者释放执行事务所需的资源,并结束这次事务的执行;
  • 完成步骤2后,参与者发送一个ack消息给协作者
  • 协作者收到所有参与者的ack消息后,标识该事务执行完成

回滚

只有在所有检查点都成功完成这个前提下,写入才会成功。这符合2PC的流程,其中JobManager为协调者,各个算子为参与者(不过只有sink一个参与者会执行提交)。一旦有检查点失败,notifyCheckpointComplete()方法就不会执行。如果重试也不成功的话,最终会回滚事务。

总结

  1. 事务开启:在sink task向下游写数据之前,均会开启一个事务,后续所有写数据的操作均在这个事务中执行,事务未提交前,事务写入的数据下游不可读;
  2. 预提交阶段: JobManager 开始下发Checkpoint Barrier,当各个处理逻辑接收到barrier后停止处理后续数据,对当前状态制作快照,此时sink也不在当前事务下继续处理数据( 处理后续的数据需要新打开下一一个事务)。状态制作成功则向JM成功的消息,失败则发送失败的消息;
  3. 提交阶段: 若JM收到所有预提交成功的消息,则向所有处理逻辑( 包括sink )发送可以提交此次事务的消息,sink 接收到此消息后,则完成此次事务的提交,此时下游可以读到这次事务写入的数据;若JM有收到预提交失败的消息,则通知所有处理逻辑回滚这次事务的操作,此时sink则丢弃这次事务提交的数据下。

标签:事务,Flink,快照,处理,制作,Exactly,算子,数据,Once
From: https://blog.51cto.com/u_12482515/7872579

相关文章

  • 搭建Flink集群、集群HA高可用以及配置历史服务器
    Flink集群搭建集群规划节点node01node02node03角色JobManager<br>TaskManagerTaskManagerTaskManager下载并解压安装包wgethttps://repo.huaweicloud.com/apache/flink/flink-1.17.0/flink-1.17.0-bin-scala_2.12.tgz在node01节点下载flink安装包,同时解......
  • Flink引擎介绍
    Flink概述——什么是FlinkApacheFlink是一个框架和分布式处理引擎,用于在无边界和有边界数据流上进行有状态的计算。Flink能在所有常见集群环境中运行,并能以内存速度和任意规模进行计算。ApacheFlink功能强大,支持开发和运行多种不同种类的应用程序。它的主要特性包括:批流一体......
  • Test class should have exactly one public zero-argument constructor(测试类应该只
    在练习重写equals方法时写测试方法遇到这个问题先放报错代码:publicclassOrder{intorderId;StringorderName;publicintgetOrderId(){returnorderId;}publicvoidsetOrderId(intorderId){this.orderId=orderId;}......
  • Flink测试利器之DataGen初探
    什么是FlinksqlFlinkSQL是基于ApacheCalcite的SQL解析器和优化器构建的,支持ANSISQL标准,允许使用标准的SQL语句来处理流式和批处理数据。通过FlinkSQL,可以以声明式的方式描述数据处理逻辑,而无需编写显式的代码。使用FlinkSQL,可以执行各种数据操作,如过滤、聚合、连......
  • Concepts in ML
    生成模型和判别模型https://blog.csdn.net/weixin_39910711/article/details/89483662Encoder-Decoder架构https://blog.51cto.com/u_15588078/6531178?u_atoken=2514a604-93d0-4edb-baf9-b27e4864212f&u_asession=01aPKTemGjj0tyVgdoeSTz541-soymXAnN7yXDtVU-CVxi_XC_y0wsqWyr......
  • flink优化
    1、时间定义、事件时间和处理时间https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/table/concepts/time_attributes/#defining-in-ddl-12、自定义函数https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/table/functions/udfs......
  • 直播平台源码,FlinkSQL实现行转列
    直播平台源码,FlinkSQL实现行转列1、使用UNNEST解析 select name,course,scorefromods_kafka_student_scores CROSSJOINUNNEST(`list`)ASt(course,score);select name,course,scorefromods_kafka_student_scores,UNNEST(`list`)ASt(course,score);select name......
  • 重新编译kyuubi-1.6.1版本使其提交flink sql流式任务时不阻塞
    kyuubi-1.6.1版本对于flinksql的支持不是很好,在提交流式任务时会阻塞进程,为了修复这个缺陷,需要修改源代码并重新编译待编译的kyuubi版本:kyuubi-1.6.1-incubating适配的flink版本:flink-1.14.41、下载kyuubi-1.6.1-incubating版本的源代码,并导入IDEA中gitclone-bv1.6.1-incu......
  • flink 与 es 的一些问题
    写入esmaping字段类型冲突error1:org.apache.flink.streaming.connectors.elasticsearch.util.NoOpFailureHandler.onFailure(NoOpFailureHandler.java:40)-FailedElasticsearchitemrequest:Elasticsearchexception[type=mapper_parsing_exception,reason=failedtop......
  • flink序列化类型验证
    flink支持的序列化类型官方支持javatuplesandscalacaseclassesjavapojosprimitivetypesregularclassesvalueshadoopwritablesspeclalTypes验证代码StreamExecutionEnvironmentenv=StreamExecutionEnvironment.getExecutionEnvironment();......