pythonflinkkafka_ApacheFlink结合Kafka构建端到端的ExactlyOnce处理

文章目录&＃xff1a;

Apache Flink 应用程序中的 Exactly-Once 语义

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

示例 Flink 应用程序启动预提交阶段

在 Flink 中实现两阶段提交 Operator

总结

Apache Flink 自2017年12月发布的1.4.0版本开始&＃xff0c;为流计算引入了一个重要的里程碑特性&＃xff1a;TwoPhaseCommitSinkFunction(相关的Jira)。它提取了两阶段提交协议的通用逻辑&＃xff0c;使得通过Flink来构建端到端的Exactly-Once程序成为可能。同时支持一些数据源(source)和输出端(sink)&＃xff0c;包括Apache Kafka 0.11及更高版本。它提供了一个抽象层&＃xff0c;用户只需要实现少数方法就能实现端到端的Exactly-Once语义。

有关TwoPhaseCommitSinkFunction的使用详见文档: TwoPhaseCommitSinkFunction。或者可以直接阅读Kafka 0.11 sink的文档: kafka。

接下来会详细分析这个新功能以及Flink的实现逻辑&＃xff0c;分为如下几点。

描述Flink checkpoint机制是如何保证Flink程序结果的Exactly-Once的

显示Flink如何通过两阶段提交协议与数据源和数据输出端交互&＃xff0c;以提供端到端的Exactly-Once保证

通过一个简单的示例&＃xff0c;了解如何使用TwoPhaseCommitSinkFunction实现Exactly-Once的文件输出

一、Apache Flink应用程序中的Exactly-Once语义

当我们说『Exactly-Once』时&＃xff0c;指的是每个输入的事件只影响最终结果一次。即使机器或软件出现故障&＃xff0c;既没有重复数据&＃xff0c;也不会丢数据。

Flink很久之前就提供了Exactly-Once语义。在过去几年中&＃xff0c;我们对Flink的checkpoint机制有过深入的描述&＃xff0c;这是Flink有能力提供Exactly-Once语义的核心。Flink文档还提供了该功能的全面概述。

在继续之前&＃xff0c;先看下对checkpoint机制的简要介绍&＃xff0c;这对理解后面的主题至关重要。

次checkpoint是以下内容的一致性快照&＃xff1a;

应用程序的当前状态

输入流的位置

Flink可以配置一个固定的时间点&＃xff0c;定期产生checkpoint&＃xff0c;将checkpoint的数据写入持久存储系统&＃xff0c;例如S3或HDFS。将checkpoint数据写入持久存储是异步发生的&＃xff0c;这意味着Flink应用程序在checkpoint过程中可以继续处理数据。

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

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

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

分布式系统中&＃xff0c;协调提交和回滚的常用方法是两阶段提交协议。在下一节中&＃xff0c;我们将讨论Flink的TwoPhaseCommitSinkFunction是如何利用两阶段提交协议来提供端到端的Exactly-Once语义。

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

我们将介绍两阶段提交协议&＃xff0c;以及它如何在一个读写Kafka的Flink程序中实现端到端的Exactly-Once语义。Kafka是一个流行的消息中间件&＃xff0c;经常与Flink一起使用。Kafka在最近的0.11版本中添加了对事务的支持。这意味着现在通过Flink读写Kafaka&＃xff0c;并提供端到端的Exactly-Once语义有了必要的支持。

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

在今天讨论的这个示例程序中&＃xff0c;我们有&＃xff1a;

从Kafka读取的数据源(Flink内置的KafkaConsumer)

窗口聚合

将数据写回Kafka的数据输出端(Flink内置的KafkaProducer)

要使数据输出端提供Exactly-Once保证&＃xff0c;它必须将所有数据通过一个事务提交给Kafka。提交捆绑了两个checkpoint之间的所有要写入的数据。这可确保在发生故障时能回滚写入的数据。但是在分布式系统中&＃xff0c;通常会有多个并发运行的写入任务的&＃xff0c;简单的提交或回滚是不够的&＃xff0c;因为所有组件必须在提交或回滚时“一致”才能确保一致的结果。Flink使用两阶段提交协议及预提交阶段来解决这个问题。

在checkpoint开始的时候&＃xff0c;即两阶段提交协议的“预提交”阶段。当checkpoint开始时&＃xff0c;Flink的JobManager会将checkpoint barrier(将数据流中的记录分为进入当前checkpoint与进入下一个checkpoint)注入数据流。

brarrier在operator之间传递。对于每一个operator&＃xff0c;它触发operator的状态快照写入到state backend。

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

这种方式仅适用于operator具有『内部』状态。所谓内部状态&＃xff0c;是指Flink state backend保存和管理的 -例如&＃xff0c;第二个operator中window聚合算出来的sum值。当一个进程有它的内部状态的时候&＃xff0c;除了在checkpoint之前需要将数据变更写入到state backend&＃xff0c;不需要在预提交阶段执行任何其他操作。Flink负责在checkpoint成功的情况下正确提交这些写入&＃xff0c;或者在出现故障时中止这些写入。

三、示例Flink应用程序启动预提交阶段

但是&＃xff0c;当进程具有『外部』状态时&＃xff0c;需要作些额外的处理。外部状态通常以写入外部系统(如Kafka)的形式出现。在这种情况下&＃xff0c;为了提供Exactly-Once保证&＃xff0c;外部系统必须支持事务&＃xff0c;这样才能和两阶段提交协议集成。

在本文示例中的数据需要写入Kafka&＃xff0c;因此数据输出端(Data Sink)有外部状态。在这种情况下&＃xff0c;在预提交阶段&＃xff0c;除了将其状态写入state backend之外&＃xff0c;数据输出端还必须预先提交其外部事务。

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

下一步是通知所有operator&＃xff0c;checkpoint已经成功了。这是两阶段提交协议的提交阶段&＃xff0c;JobManager为应用程序中的每个operator发出checkpoint已完成的回调。

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

我们对上述知识点总结下&＃xff1a;

旦所有operator完成预提交&＃xff0c;就提交一个commit。

如果至少有一个预提交失败&＃xff0c;则所有其他提交都将中止&＃xff0c;我们将回滚到上一个成功完成的checkpoint。

在预提交成功之后&＃xff0c;提交的commit需要保证最终成功 – operator和外部系统都需要保障这点。如果commit失败(例如&＃xff0c;由于间歇性网络问题)&＃xff0c;整个Flink应用程序将失败&＃xff0c;应用程序将根据用户的重启策略重新启动&＃xff0c;还会尝试再提交。这个过程至关重要&＃xff0c;因为如果commit最终没有成功&＃xff0c;将会导致数据丢失。

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

四、在Flink中实现两阶段提交Operator

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

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

beginTransaction – 在事务开始前&＃xff0c;我们在目标文件系统的临时目录中创建一个临时文件。随后&＃xff0c;我们可以在处理数据时将数据写入此文件。

preCommit – 在预提交阶段&＃xff0c;我们刷新文件到存储&＃xff0c;关闭文件&＃xff0c;不再重新写入。我们还将为属于下一个checkpoint的任何后续文件写入启动一个新的事务。

commit – 在提交阶段&＃xff0c;我们将预提交阶段的文件原子地移动到真正的目标目录。需要注意的是&＃xff0c;这会增加输出数据可见性的延迟。

abort – 在中止阶段&＃xff0c;我们删除临时文件。

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

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

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

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

总结

总结下本文涉及的一些要点&＃xff1a;

Flink的checkpoint机制是支持两阶段提交协议并提供端到端的Exactly-Once语义的基础。

这个方案的优点是: Flink不像其他一些系统那样&＃xff0c;通过网络传输存储数据 – 不需要像大多数批处理程序那样将计算的每个阶段写入磁盘。

Flink的TwoPhaseCommitSinkFunction提取了两阶段提交协议的通用逻辑&＃xff0c;基于此将Flink和支持事务的外部系统结合&＃xff0c;构建端到端的Exactly-Once成为可能。

从Flink 1.4.0开始&＃xff0c;Pravega和Kafka 0.11 producer都提供了Exactly-Once语义&＃xff1b;Kafka在0.11版本首次引入了事务&＃xff0c;为在Flink程序中使用Kafka producer提供Exactly-Once语义提供了可能性。

Kafaka 0.11 producer的事务是在TwoPhaseCommitSinkFunction基础上实现的&＃xff0c;和at-least-once producer相比只增加了非常低的开销。

这是个令人兴奋的功能&＃xff0c;期待Flink TwoPhaseCommitSinkFunction在未来支持更多的数据接收端。

via&＃xff1a;https://www.ververica.com/blog/end-to-end-exactly-once-processing-apache-flink-apache-kafka

作者&＃xff1a;Piotr Nowojski

翻译| 周凯波

周凯波&＃xff0c;阿里巴巴技术专家&＃xff0c;四川大学硕士&＃xff0c;2010年毕业后加入阿里搜索事业部&＃xff0c;从事搜索离线平台的研发工作&＃xff0c;参与将搜索后台数据处理架构从MapReduce到Flink的重构。目前在阿里计算平台事业部&＃xff0c;专注于基于Flink的一站式计算平台的建设。

本文作者&＃xff1a;apache_flink

本文为云栖社区原创内容&＃xff0c;未经允许不得转载。