vlambda博客
学习文章列表

中介绍了Flink架构，算子，Task/subtask以及Slot等基础内容，本文将继续介绍状态，窗口，水位线，checkpoint等相关内容。

01 状态

许多场景只是独立的检测流中的每一个事件（如事件解析器），但一些场景需要记录跨多个事件的信息（如window），称为状态化。 有状态的计算是Flink框架实现的重要功能，稍微复杂的流处理场景都需要记录状态，然后在新流入数据的基础上不断更新状态。

• 检测流的某种特定模式，需要缓存当前已经流入事件序列

• 按分/小时/天聚合事件，保持当前的聚合状态，增量聚合

• 机器学习场景下，需要根据新流入数据不断更新模型参数

  
  

在 一文中我们知道，Flink的一个算子有多个子任务，每个子任务分布在不同实例上（slot）， 可以把状态理解为某个算子子任务在其当前实例上的一个变量，变量记录了数据流的历史信息，当新数据流入时，我们可以结合历史信息来进行计算并更新状态。

托管状态（Managed State）

Flink托管，自动存储、恢复和伸缩，支持常见的数据结构，如ValueState、ListState、MapState等，Keyed State和Operator State都是托管状态。

原生状态（Raw State）

应用自己管理，只支持字节数组。

Keyed State

• 只能用于KeyedStream，如keyBy（）算子之后

• 相同的Key，最终映射到同一个subtask（如上图，所有的k1均映射到slot1），不同的Key可能映射到同一个subtask（slot1存在k1，k2）
  

• State生命周期存在于subtask（算子）实例内，不同的subtask实例即便是相同的Key和同名的State，Value也是不相同的；也就是说State是subtask隔离的，并不是贯穿事件的整个流处理周期

• Rescale

1. Keyed State Rescale是基于KeyGroup，分组的数量等于maxParallelism，因此maxParallelism不能变化，maxParallelism变化会引起KeyGroup发生变化，导致数据无法恢复

2. Rescale是由于算子parallesism变化引起，重新计算映射的slot即可
   

Operator State

Operator State可以用在所有算子上， Operator State只和并行的算子实例绑定，与记录中的Key无关。和Keyed State类似，每个算子的subtask或者说每个算子实例共享一个状态，但算子subtask之间的状态不能互相访问。相比Keyed State，Operator State不常用，主要用于Source，Sink记录位置信息。

Rescale

• ListState收集所有subtask的状态，然后均分给新的subtask
  

• UnionState将所有的状态都分配给新的每个subtask
  

状态存储

• MemoryStateBackend，状态存储在JVM堆内存，checkpoint时存储到Job Manager内存

• FsStateBackend，状态存储在JVM堆内存，checkpoint时存储到文件系统，可以是本地文件系统或者其他文件系统，如HDFS

• RocksDBStateBackend，状态存储在RocksDB中，checkpoint时存储到文件系统，可以是本地文件系统或者其他文件系统，如HDFS
  

使用流程

• 定义状态描述符并初始化

• 如果是重建，则恢复状态（算子状态）

• 更新状态调用update

• 快照事件发生时调用snapshot保存状态（算子状态）
  

状态相关的主要逻辑有两个

• 将算子subtask本地内存数据在checkpoint时snapshot写入存储

• 初始化或重启应用时，从存储中读出，恢复算子subtask的本地内存数据

Keyed State对这两项内容做了更完善的封装，开发者无需关心。对于Operator State来说需要根据自己的业务场景来设计snapshot和restore的逻辑，Flink提供了CheckpointFunction接口。

---

02 checkpoint

checkpoint是Flink容错的核心机制，它可以定期将各个算子的状态数据进行快照存储（snapshot）。如果 Flink程序出现宕机，可以重新从这些快照中恢复数据。

• barrier是分布式快照的核心，它被插入到数据流中作为流的一部分，并与记录一起流动

• barrier不会干扰正常的记录，barrier将记录切分为snapshot集合
  

• barrier携带了一个快照ID，它非常轻量级，不会影响流的性能

• 一次快照贯穿从source到sink整个流程。Job Manager产生barrier n并向下游传递，中间算子subtask收到来自上游算子subtask（可能有多个）的全部barrier n，进行状态snapshot（异步持久化），然后向下游算子subtask（可能有多个）广播barrier n，一旦有一个sink接收到来自上游算子subtask的全部barrier n，就会向checkpoint协调器（Job Manager）标记snapshot n，当所有的sink都标记了snapshot n，则认为snapshot n完成

  
  

  
  

carrier对齐

• 当上游算子有多个输入时，多个输入中的carrier会出现到达时间不一致问题。等到上游输入所有的carrier到达以后，再进行snapshot称为carrier对齐

• 如上图，第一个输入流carrier到达以后，不再处理这个输入流的记录，后续输入放入缓冲区（input buffer）

• 第二个输入流正常处理，当第二个输入流的carrier到达，触发checkpoint：

1. 向下游广播carrier

2. snapshot当前算子

3. 处理缓冲区数据

4. 继续处理流中的数据

优点

保存数据少， Exactly Once

缺点

延时高，checkpoint时间长，加剧反压（在短时内的负载高峰导致系统接收数据速率远大于它能处理数据的速率）

carrier不对齐

• 不需要等到上游输入所有的carrier到达，一旦有carrier到达（第一个），就将carrier插入到输出流的最前端，然后进行snapshot称为carrier不对齐

优点

延时低，checkpoint时间短，不会加剧反压，At Least Once（当第一个carrier n到达后，算子持续处理后续carrier n+1的记录，宕机从carrier n恢复时，已经处理的carrier n+1的记录将被重复处理）

缺点

保存数据多，恢复比较慢

---

03 窗口

有界流

有起始有结束，可以提前获取全部数据，吞吐量大，对应批处理

无界流

有开始无结束，持续产生，即刻处理，对应流处理

Flink认为数据均是流式的，批处理只是流处理的特例，对于批量和流处理分别对应DataSet API和DataStream API。

窗口是处理无限流数据的核心，它将流切分为有限大小buckets（桶），所有的计算都在桶内进行，窗口是从流处理到批量处理的一个桥梁。

每个窗口都有开始时间和结束时间，属于窗口的第一个元素到达，就会创建窗口，当时间（事件或处理时间）超过其结束时间戳加上用户指定的允许延迟时，触发计算，然后窗口将被关闭（清空）

关于时间

• Event Time，事件在现实世界中发生的时间，它通常由事件中的时间戳描述

• Ingestion Time，数据进入Flink流处理系统的时间，也就是Flink读取数据源时间

• Processing Time，数据流入到具体某个subtask算子 (消息被计算处理) 时候相应的系统时间，也就是Flink程序处理该事件时当前系统时间

翻滚窗口（Tumbling）

固定大小的窗口，滚动窗口大小固定的并且不重叠

滚动窗口（Sliding）

滑动窗口与滚动窗口的区别是滑动窗口有重复的计算部分，重叠大小由滑动步长的大小决定

---

04 水位线（Watermark）

水位线是Flink为了处理Event Time窗口计算提出的一种机制， 用于处理乱序事件或延迟数据的，本质上也是一种时间戳。 通常水位线与窗口结合使用， 比如对于延迟的数据，系统不能无限期的等待下去，需要有个机制来保证一个特定的时间后，必须触发window进行计算。

W(T)的语义是T时间之前的事件都已经到达

触发（计算）条件：

   W(T) >= 窗口结束时间，且[窗口开始时间，窗口结束时间)有数据

设置水位线

通过在DataStream上调用assignTimestampsAndWatermarks方法来分配时间和水位线，两种参数AssignerWithPeriodicWatermarks和AssignerWithPunctuatedWatermarks

• 标点水位线（AssignerWithPunctuatedWatermarks）

  通过流中某些特殊的标志事件来触发新水位线的生成（如事件的时间戳timestamp%10==0生成新水位线）。这种方式下窗口的触发与时间无关，而是决定于何时收到标志事件。这种情况下可以为每个事件都生成一个水位线，但水位线需参与计算，设置过多的水位线会导致整体计算性能下降，只有在实时性要求非常高的场景才会选择标点水位线

  
  

• 定期水位线（AssignerWithPeriodicWatermarks）

  周期性的（一定时间间隔或者达到一定的记录条数）产生一个Watermark，在实际的生产中，定期水位线通常需要结合时间和积累条数两个维度周期性产生Watermark，否则在极端情况下会有很大的延时

迟到事件

也就是在窗口关闭之后到达的时间，Flink有3种处理方式

• 丢弃（默认）

  
  

• allowedLateness

  在Watermark允许延迟的基础上，再增加一个延迟时间。实际事件允许延迟的时间等于水位线允许的延迟时间+allowedLateness。Flink会在窗口关闭后一直保存窗口的状态直至超过允许迟到时长，这期间的迟到事件不会被丢弃，而是默认会触发窗口重新计算。

  
  

• sideOutputLateData

  兜底操作，将迟到事件单独放入一个数据流分支，作为window计算结果的副产品，以便对其进行特殊处理