【Flink】介绍Flink中状态一致性的保证

你的名字 2022-09-07 09:54 175阅读 0赞

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70]

# 1.概述 #

转载：[介绍Flink中状态一致性的保证][Flink] 再次温习了这篇文章有了不一样的收货。侵权可删，这里是方便自己找到。

## 1. 一致性 ##

### 1.1 介绍状态一致性 ###

*  有状态的流处理，内部每个算子任务都可以有自己的状态
 *  对于流处理器内部来说，所谓的状态一致性，其实就是我们所说的要保证计算结果准确，一条数据有也不丢失，也不会重复计算数据
 *  在程序遇到故障时可以恢复任务状态，恢复以后的任务重新计算数据，计算完的结果也应该是完全正确的

### 1.2 性级别有哪些 ###

在流处理中，一致性可以分为 3 个级别：

<table> 
 <thead> 
  <tr> 
   <th>级别</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><strong>at-most-once</strong><br>至多一次</td> 
   <td>最多处理一次，当任务发生故障时，什么都不做，<br>既不恢复丢失的状态，也不重播丢失的数据。<br>这其实是没有正确性保障的委婉说法——故障发生之后， 计数结果可能丢失。同样的还有 udp。</td> 
  </tr> 
  <tr> 
   <td><strong>at-least-once</strong><br>至少一次</td> 
   <td>最少处理一次，所有的事件都会得到处理，<br>这表示计数结果可能大于正确值，但绝不会小于正确值。<br>也就是说，计数程序在发生故障后可能多算，但是绝不会少算。</td> 
  </tr> 
  <tr> 
   <td><strong>exactly-once</strong><br>精确一次</td> 
   <td>每个事件都会被处理且仅会被处理一次，这指的是系统保证在发生故障后得到的计数结果与正确值一致。</td> 
  </tr> 
 </tbody> 
</table>

曾经， at-least-once 非常流行。第一代流处理器(如 Storm 和 Samza)刚问世时只保证 at-least-once，原因有二。

1.  保证 exactly-once 的系统实现起来更复杂。这在基础架构层(决定什么代表正确，以及 exactly-once 的范围是什么)和实现层都很有挑战性。
2.  流处理系统的早期用户愿意接受框架的局限性，并在应用层想办法弥补(例如使应用程序具有幂等性，或者用批量计算层再做一遍计算)。

最先保证 exactly-once 的系统(Storm Trident 和 Spark Streaming)在性能和表现力这两个方面付出了很大的代价。为了保证 exactly-once，这些系统无法单独地对每条记录运用应用逻辑，而是同时处理多条(一批)记录，保证对每一批的处理要么全部 成功，要么全部失败。这就导致在得到结果前，必须等待一批记录处理结束。因此， 用户经常不得不使用两个流处理框架(一个用来保证 exactly-once，另一个用来对每个元素做低延迟处理)，结果使基础设施更加复杂。曾经， 用户不得不在保证exactly-once 与获得低延迟和效率之间权衡利弊。Flink 避免了这种权衡。

Flink 的一个重大价值在于， 它既保证了 exactly-once， 也具有低延迟和高吞吐的处理能力。从根本上说，Flink 通过使自身满足所有需求来避免权衡，它是业界的一次意义重大的技术飞跃。尽管这在外行看来很神奇，但是一旦了解，就会恍然大悟。

### 1.3 一致性检查点 ###

Flink 使用了一种轻量级快照机制 — 检查点（checkpoint） 来保证 exactly-once 语义

有状态应用的一致性检查点，就是所有任务的装在，在某个时间点的一份快照，而这个时间点，应该是所有任务都恰好处理完一个相同的输入数据的时候，

应用状态的一致检查点，就是 flink 故障恢复机制的核心

## 2. 端到端的一致性 ##

*  端到端（end-to-end）的状态一致性意味着从数据来源的 source 到转换算子再到 sink 能够有一致性保证
 *  这意味着结果的正确性贯穿整个流处理应用的始终每个组件都保证了它自己的一致性。
 *  整个端到端的一致性级别取决去所有组件中一致性最弱的组件

### 2.1 端到端 exactly-once 各部分的实现方式 ###

*  内部保证 - checkpoint，发生故障时能够恢复各个环节的数据
 *  source端 - 可重设数据的读取位置，当发生故障时重置偏移量到故障之前的位置
 *  sink 端 - 从故障恢复时，数据不会重复写入外部系统
    
     *  幂等写入
     *  事务写入

### 2.2 幂等写入（idempotent Writes） ###

所谓幂等操作，是说一个操作，可以重复执行很多次，但只导致一次结果更改，也就是说后面再重复执行就不起作用了。

### 2.3 事务写入(Transactional Writes) ###

*  事务（Transaction）
    
     *  应用程序中一系列严密的操作，所有操作必须成功完成，否则在每个操作中所有的所有更改都会被撤销
     *  具有原子性：一个事务中的一系列的操作要么全部成功，要么一个都不做
 *  实现思想：构建的事务对应着 checkpoint，等到 checkpoint 真正完成的时候，才把所有对应的结果写入 sink 系统中
 *  实现方式
    
     *  预写日志
     *  两阶段提交

#### 2.3.1 预写日志 ####

*  Write-Ahead-Log, WAL
 *  把结果数据先当成状态保存，然后再收到 checkpoint 完成的通知时，一次性写入 sink 系统
 *  简单易于实现，由于数据提前在状态后端做了缓存，所以无论什么 sink 系统，都能用这种方式一批搞定
 *  存在的问题：写入数据时出现故障则会导致一部分数据成功一部分失败
 *  DataStream API 提供了一个模板类：GenericWriteAheadSink，来实现这种事务性 sink

#### 2.3.2 两阶段提交 ####

*  Two-Phase-Commit， 2PC
 *  对于每个 checkpoint，sink 任务会启动一个事务，并将接下来所有接收的数据添加到事务里
 *  然后将这些数据写入外部 sink，但不提交它们，这时只是“预提交”
 *  当它收到 checkpoint 完成的通知时，它才正式提交事务，实现结果的真正写入
 *  这种方式真正实现了 exactly-once，它需要一个提供事务支持的外部 sink 系统，Flink 提供了 TwoPhaseCommitSinkFunction 接口

#### 2.3.3 pc 对外部sink 系统的要求 ####

*  外部 sink 系统必须提供事务支持，或者 sink 任务必须能够模拟外部系统上的事务
 *  在 checkpoint 的隔离期间里，必须能够开启一个事务并接受数据写入
 *  在收到 checkpoint 完成的通知之前，事务必须是“等待提交”的状态。在故障恢复的情况下，这可能需要一些时间。如果这个时候 sink 系统关闭事务（例如超时了），那么未提交的数据就会丢失
 *  sink 任务必须能够在进程失败后恢复事务
 *  提交事务必须是幂等操作

### 2.4 Source 和 Sink 的一致性保证 ###

<table> 
 <thead> 
  <tr> 
   <th>sink\source</th> 
   <th>不可重置</th> 
   <th>可重置</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>任意（Any）</td> 
   <td>at-most-once</td> 
   <td>at-least-once</td> 
  </tr> 
  <tr> 
   <td>幂等</td> 
   <td>at-most-once</td> 
   <td>exactly-once<br>故障恢复时会出现暂时不一致</td> 
  </tr> 
  <tr> 
   <td>预写日志（WAL）</td> 
   <td>at-most-once</td> 
   <td>at-least-once</td> 
  </tr> 
  <tr> 
   <td>两阶段提交（2PC）</td> 
   <td>at-most-once</td> 
   <td>exactly-once</td> 
  </tr> 
 </tbody> 
</table>

## 3. Flink + Kafka 的端到端状态一致性保证 ##

1.  **flink内部**：利用 checkpoint 机制把状态保存，当发生故障的时候可以恢复状态，从而保证内部的状态一致性
2.  **source 端**：kafka consumer 作为 source，可以将偏移量保存下来，当发生故障时可以从发生故障前的偏移量重新消费数据，从而保证一致性
3.  **sink端**：kafka producer 作为 sink，采用两阶段提交 sink，需要实现一个 TwoPhaseCOmmitSinkFunction

### 3.1 Exactly-once 两阶段提交 ###

1.  JobManager 协调各个 TaskManager 进行 checkpoint 存储，Checkpoint 保存在 StateBackend 中，默认 StateBackend 是内存级别，也可以改为文件级进行持久化保存

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 1]

1.  当 checkpoint 启动时，JobManager 会将 barrier 注入到数据流中，在 source 读取到 barrier 时会进行检查点操作，执行完检查点操作后将 barrier 下发到下游算子。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 2]

1.  当算子收到 barrier 时，会将当前保存到状态后端，checkpoint 可以保证内部的状态一致性。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 3]

1.  每个内部的任务遇见barrier 时，都会把状态保存到checkpoint 里，sink 任务首先把数据写入外部 kafka，这些事务都属于预提交事务，遇到 barrier 时把状态保存到状态后端，并开启新的预提交事务。barrier 之前的数据属于上一个事务，barrier 之后的数据属于下一个事务，

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 4]

1.  当所有环节的 checkpoint完成时，JobManager 会向所有任务发出通知，确认这次 checkpoint 完成，当 sink 收到这个完成的通知时，正式提交之前的事务，kafka 中未确认的数据状态改为已确认。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 5]

### 3.2 两阶段提交步骤 ###

具体的两阶段提交步骤总结如下：

1.  sink收到第一条数据之后，开启一个 kafka 的事务（ transaction），这时把数据写入kafka 标记为未提交， 这就是“预提交”
2.  jobmanager 触发 checkpoint 操作，barrier 从 source 开始向下传递，遇到barrier 的算子将状态存入状态后端，并通知 jobmanager
3.  sink 连接器收到 barrier，保存当前状态，存入 checkpoint，通知jobmanager，并开启下一阶段的事务，用于提交下个检查点的数据
4.  jobmanager 收到所有任务的通知，发出确认信息，表示 checkpoint 完成
5.  sink 任务收到 jobmanager 的确认信息，正式提交这段时间的数据
6.  外部 kafka 关闭事务，提交的数据可以正常消费了。

所以我们也可以看到，如果宕机需要通过 StateBackend 进行恢复， 只能恢复所有确认提交的

### 3.3 注意点 ###

1.  kafka超时时间要与 sink 超时要保持一致
2.  kafka 设置未提交数据不可读

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70]: /images/20220829/aec29eb3f47a4297a5ef00e67263cae9.png
[Flink]: https://www.hnbian.cn/posts/6adf75db.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 1]: /images/20220829/e2ccadd8eb0f4261a96d5d576c78b7ab.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 2]: /images/20220829/49d2ffb2afb14688a1431efbb22ed075.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 3]: /images/20220829/4f06af3c039e4752977c13717dbdf0f8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 4]: /images/20220829/d923baaa4639465aa3c63e55b642a772.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 5]: /images/20220829/9a8489d65f144aea90f9d014e7700822.png