一文搞定 Flink Task 提交执行全流程

Posted 2023-04-25

参考技术A 上一篇 一文搞定 Flink Job 提交全流程 ，我们知道每一个 operator chain 作为一个整体，提交 task 。

这里创建了一个 Task 对象并启动，我们来看一下 Task 启动的时候都做了什么

首先向 blobService、netWork 注册 job ，添加监控，将jar 等添加到分布式缓存中，然后就 invoke，这也是 task 真正开始执行的地方，我们以 StreamTask 为例

init 然后对于一些 Rich Function 会先执行其 open方法，然后开始 run，就开始真正的消费数据了。我们以 flatMap 为例
当执行 run 方法时，首先呢 OneInputStreamTask.run

这一块的逻辑可具体参考
一文搞定 Flink 消费消息的全流程 、 一文搞定 Flink Checkpoint Barrier 全流程 以及 一文搞懂 Flink 处理 Barrier 全过程
我们知道当往下游发送数据的时候

继续追踪下去到 StreamFlatMap.processElement

其他的类似，如果是 kafka source task 调用的是 headOperator.run(getCheckpointLock(), getStreamStatusMaintainer())，然后去消费 kafka 中的数据。然后跟前面 一文搞定 Flink Job 提交全流程 、 写给大忙人看的Flink 消费 Kafka 、 一文搞定 Flink 消费消息的全流程 以及 一文搞定 Flink Checkpoint Barrier 全流程 就可以串起来了。而 Flink 整体流程的分析，除了 restore 之外，也差不多可以告一段落了。

4.一文搞定：Flink与Kafka之间的精准一次性

参考技术A 在上一篇文章当中，也算是比较详细且通俗的聊了聊Flink是如何通过checkpoint机制来完成数据精准一次性的实现的。并且也在上一章的结尾表示，要在接下来聊一聊Flink与它的铁哥们Kafaka之间，是如何实现数据的精准一次性消费的。
本次的聊法，还是要通过以kafka（source）->Flink,Flink(source)->Kafka来分别展开讨论。

kafka是一个具有数据保存、数据回放能力的消息队列，说白了就是kafka中的每一个数据，都有一个专门的标记作为标识。而在Flink消费kafka传入的数据的时候，source任务就能够将这个偏移量以算子状态的角色进行保存，写入到设定好的检查点中。这样一旦发生故障，Flink中的FlinkKafkaProduce连接器就i能够按照自己保存的偏移量，自己去Kafka中重新拉取数据，也正是通过这种方式，就能够确保Kafka到Flink之间的精准一次性。

在上一篇文章当中，已经表明了，如果想要让输出端能够进行精准一次性消费，就需要使用到幂等性或者是事务。而事务中的两阶段提交是所有方案里面最好的实现。

其实Flink到Kafak之间也是采用了这种方式，具体的可以看一下ctrl进到FlinkKafkaProduce连接器内部去看一看：

这也就表明了，当数据通过Flink发送给sink端Kafka的时候，是经历了两个阶段的处理的。第一阶段就是Flink向Kafka中插入数据，进入预提交阶段。当JobManager发送的Ckeckpoint保存成功信号过来之后，才会提交事务进行正式的数据发送，也就是让原来不可用的数据可以被使用了。

这个实现过程到目前阶段就很清晰了，它的主体流程无非就是在开启检查点之后，由JobManager向各个阶段的处理逻辑发送有关于检查点的barrier。所有的计算任务接收到之后，就会根据自己当前的状态做一个检查点保存。而当这个barrier来到sink任务的时候，sink就会开启一个事务，然后通过这个事务向外预写数据。直到Jobmanager来告诉它这一次的检查点已经保存完成了，sink就会进行第二次提交，数据也就算是成功写出了。

1.必须要保证检查点被打开了，如果检查点没有打开，那么之前说的一切话都是空谈。因为Flink默认检查点是关着的。
2.在FlinkKafakProducer连接器的构造函数中要传入参数，这个参数就是用来保证状态一致性的。就是在构造函数的最后一个参数输入如下：

3.配置Kafka读取数据的隔离级别
在kafka中有个配置，这个配置用来管理Kafka读取数据的级别。而这个配置默认是能够读取预提交阶段的数据的，所以如果你没改这个配置，那两阶段提交的第一阶段就是白费了。所以需要改一下这个配置，来更换一下隔离级别：

4.事务超时时间
这个配置也很有意思，大家试想一下。如果要进行两阶段提交，就要保证sink端支持事务，Kafka是支持事务的，但是像这个组件对于很多机制都有一个超时时间的概念，也就是说如果时间到了这个界限还没完成工作，那就会默认这个工作失败。Kafka中由这个概念，Flink中同样由这个概念。但是flink默认超时时间是1小时，而Kafka默认是15分钟，这就有可能出现检查点保存东西的时间大于15分钟，假如说是16分钟保存完成然后给sink发送检查点保存陈功可以提交事务的信号，但是这个时候Kafka已经认为事务失败，把之前的数据都扔了。那数据不就是丢失了么。所以说Kafka的超时时间要大于Flink的超时时间才好。

截止到目前为止，基本上把有关于状态维护的一些东西都说完了，有状态后端、有检查点。还通过检查点完成可端到端的数据精准一次性消费。但是想到这我又感觉，如果有学习进度比我差一些的，万一没办法很好的理解怎么办。所以在下一篇文章当中我就聊聊Flink中的“状态”到底是个什么东西，都有什么类型，都怎么去用。