Flink

事件时间（Event Time）

摄入时间（Ingestion Time）

处理时间（Processing Time）

1、在流处理应用中，数据是连续不断的，因此我们不可能等到所有数据都到了才开始处理。当然我们可以每来一个消息就处理一次，但是有时我们需要做一些聚合类的处理，例如：在过去的1分钟内有多少用户点击了我们的网页。在这种情况下，我们必须定义一个窗口，用来收集最近一分钟内的数据，并对这个窗口内的数据进行计算。
2、Windows是flink处理无限流的核心，Windows将流拆分为有限大小的“桶”，我们可以在其上应用计算。Flink认为Batch是Streaming的一个特例，所以flink底层引擎是一个流式引擎，在上面实现了流处理和批处理。而窗口就是从Streaming到Batch的一个桥梁。
3、窗口分类：基于时间划分驱动。例如：每30秒钟
                        基于数据数量驱动。例如：每100个元素，与时间无关

keyed

non-keyed

1、计数窗口基于元素的个数来截取数据，到达固定的个数时就触发计算并关闭窗口
2、相当于座位有限、“人满就发车”，是否发车与时间无关。每个窗口截取数据的个数，就是窗口的大小。
3、计数窗口相比时间窗口就更加简单，我们只需指定窗口大小，就可以把数据分配到对应的窗口中了。

1、滚动窗口的assigner分发元素到指定大小的窗口。滚动窗口的大小时固定的，且各自范围之间不重叠。
2、滚动计数窗口只需要传入一个长整型的参数size，表示窗口的大小
3、使用场景：适用于按照指定的周期来统计指标，比如说，每两个事件做一次统计

1、滑动窗口的assigner分发元素到指定大小的窗口，窗口大小通过window size参数设置。
2、滑动窗口需要传入两个参数：size和slide，前者表示窗口大小，后者表示滑动步长。
3、如果slide小于窗口大小，滑动窗口可以允许窗口重叠。在这种情况下，一个元素可能会被分发到多个窗口
4、例如：每5个事件做一次统计，每次滑动长度2

coutWindowAll数量窗口（不区分数量滚动窗口【滑动窗口与滚动窗口的区别，在于滑动窗口会有元素重叠可能，而滚动窗口不存在元素重叠】）

1、会话窗口的assigner会把数据按活跃的会话分组
2、与滚动窗口和滑动窗口不同，会话窗口不会相互重叠，且没有固定的开始或结束时间
3、会话窗口的assigner可以设置固定的会话间隔（session gap）或用session gap extractor函数来动态地定义多长时间算作不活跃
4、当超出了不活跃地时间段，当前地会话就会关闭，并且将接下来地数据分发到新的会话窗口

1、窗口将数据收集起来，最基本的处理操作当然就是进行聚合。窗口对无限流的切分，可以看作得到了一个有界数据集。如果我们等到所有数据都收集齐，在窗口到了结束时间要输出结果的一瞬间再去进行聚合，显然就不够高效
2、为了提高实时性，我们可以再次将流处理的思路发扬光大：就像 DataStream 的简单聚合一样，每来一条数据就立即进行计算，中间只要保持一个简单的聚合状态就可以了；区别只是在于不立即输出结果，而是要等到窗口结束时间。等到窗口到了结束时间需要输出计算结果的时候，我们只需要拿出之前聚合的状态直接输出，这无疑就大大提高了程序运行的效率和实时性。

1、全量窗口函数需要对所有进入该窗口的数据进行缓存，等到窗口触发时才会遍历窗口内所有数据，进行结果计算。
2、因为有些场景下，我们要做的计算必须基于全部的数据才有效，这时做增量聚合就没什么意义了；另外，输出的结果有可能要包含上下文中的一些信息（比如窗口的起始时间），这是增量聚合函数做不到的。所以，我们还需要有更丰富的窗口计算方式，这就可以用全窗口函数来实现。

水位线就是用来度量事件事件
水位线是数据流中的一部分，随着数据一起流动，在不同任务之间传输
水位线可以看作是插入到数据流中的一个标记点，主要内容就是一个时间戳，用来指示当前的事件时间
水位线插入流中的位置，就应该是在某个数据到来之后；这样就可以从这个数据中提取时间戳，作为当前水位线的时间戳了。

1.10版本之前

1.11版本以后

周期性水位线

定点式水位线

水位线传递

现实中很难生成一个完美的水位线，水位线就是在延迟与准确性之前做的一种权衡。如果生成的水位线过于紧迫，即水位线可能会大于后来数据的时间戳，这就意味着数据有延迟，关于延迟数据的处理，Flink提供了一些机制，具体如下：allowedLateness、sideOutputLateData

flink提供了WindowedStream.allowedLateness（）方法来设定窗口的允许延迟
正常情况下窗口触发计算完成之后就会被销毁，但是设定了允许延迟之后，窗口会等待allowedLateness的时长后开始计算并销毁。在该区间内的迟到数据仍然可以进入窗口中，并触发新的计算
什么情况下数据会被丢弃或者说不会被计算？
未设置allowedLateness情况下，某条数据属于某个窗口，但是watermark超过了窗口的结
束时间，则该条数据会被丢弃；
设置allowedLateness情况下，某条数据属于某个窗口，但是watermark超过了窗口的结束
时间+延迟时间，则该条数据会被丢弃；

保底方案，数据延迟严重，可以保证数据不丢失
延迟的数据通过outputTag输出，必须要事件时间大于watermark+allowed lateness，数据才会存储在outputTag中

无状态算子

有状态算子

为什么需要状态

原始状态

托管状态

算子状态

键分区状态

简述

flink1.13版以前

flink1.13版本以后

基本概念

相关参数

对于两条流的合并，很多情况我们不是简单地将所有数据放在一起，而是希望根据某个字段的值将它们联结起来，“配对”去做处理

Tumbling Window Join

Sliding Window Join

Session Window Join

除了输出匹配的元素以外，未能匹配的元素也会输出
用于DataStream时返回是CoGroupStreams，用于DataSet时返回时CoGroupOperatorSets

业务需求

实现原理

代码实现

flink是一个stateful（带状态）的数据处理系统；系统在处理数据的过程中，各算子所记录的状态会随着数据的处理而不断变化
一旦系统崩溃，需要重启后能够恢复出崩溃前的状态才能进行数据的继续处理
因此，必须要一种机制能够对系统内的各种状态进行持久化容错。
Flink-EOS：Exactly-Once Semantics
指端到端的一致性，从数据读取，引擎计算，写入外部存储的整个过程中，即使机器或软件出现故障，都确保数据仅处理一次，不会重复，也不会丢失
一条（或者一批）数据，从注入系统，中间处理，到输出结果的整个流程中，要么每个环节都处理成功，要么失败回滚。

有可能会有数据丢失
这本质上是最简单的恢复方式，也就是直接从失败处的下个数据开始恢复程序，之前的失败数据处理就不管了。可以保证数据或事件最多由应用程序中的所有算子处理一次，这意味着如果数据在被流应用程序完全处理之前发生丢失，则不会进行其他重试或者重新发送

有可能重复处理数据
应用程序中的所有算子都保证数据或事件至少被处理一次。这通常意味着如果事件在流应用程序完全处理之前丢失，则将从源头重放或重新传输事件。然而，由于事件是可以被重传的，因此一个事件有时会被处理多次，至于有没有重复数据，不会关心，所以这种场景需要人工干预自己处理重复数据

每一条消息只被流处理系统处理一次
即使是在各种故障的情况下，流应用程序中的所有算子都保证事件只会被精确一次处理。
flink实现精确一次的分布式快照/状态检查点方法受到Chandy-Lamport分布式快照算法的启发。
流应用程序中每个算子的所有状态都会定期做checkpoint
如果是在系统中的任何地方发生失败，每个算子的所有状态都回滚到最新的全局一致checkpoint点
在回滚期间，将暂停所有处理，源也会重置为与最近checkpoint相对应的正确偏移量。
整个流应用程序基本上是回到最近一次的一致状态，然后程序从该状态重新启动。

端到端的精确一次
flink在1.4.0版本引入Exactly-Once,并号称支持端到端精确一次语义。它指的是flink应用从source端开始到sink端结束，数据必须经过的起始点和结束点

数据分割

barrier对齐

任务启动

启动checkpoint

source端

Transformation端

barrier对齐

缓存数据

Sink端

机制革新

Savepoint 作为实时任务的全局镜像，其在底层使用的代码和Checkpoint的代码是一样的
Savepoint 是依据 Flink checkpointing 机制所创建的流作业执行状态的一致镜像；
Checkpoint 的主要目的是为意外失败的作业提供恢复机制(如 tm/jm 进程挂了)。
Checkpoint 的生命周期由 Flink 管理，即 Flink 创建，管理和删除 Checkpoint - 无需用户交
互。
Savepoint 由用户创建，拥有和删除。 他们的用例是计划的，手动备份和恢复。
Savepoint 应用场景，升级 Flink 版本，调整用户逻辑，改变并行度，以及进行红蓝部署等。
Savepoint 更多地关注可移植性
Savepoint触发方式触发方式目前有三种
使用 flink savepoint 命令触发 Savepoint,其是在程序运行期间触发 savepoint。
使用 flink cancel -s 命令，取消作业时，并触发 Savepoint。
使用 Rest API 触发 Savepoint，格式为：*/jobs/:jobid /savepoints*
Savepoint注意点
由于 Savepoint 是程序的全局状态，对于某些状态很大的实时任务，当我们触发
Savepoint，可能会对运行着的实时任务产生影响，个人建议如果对于状态过大的实时任务，
触发 Savepoint 的时间，不要太过频繁。根据状态的大小，适当的设置触发时间。
当我们从 Savepoint 进行恢复时，需要检查这次 Savepoint 目录文件是否可用。可能存在你
上次触发 Savepoint 没有成功，导致 HDFS 目录上面 Savepoint 文件不可用或者缺少数据文
件等，这种情况下，如果在指定损坏的 Savepoint 的状态目录进行状态恢复，任务会启动不
起来。

事务写入

预写日志

两阶段提交，顾名思义，即分两个阶段commit：preCommit和Commit。
preCommit阶段
协调者向所有参与者发起请求，询问是否可以执行提交操作，并开始等待所有参与者的
响应。
所有参与者节点执行协调者询问发起为止的所有事务操作，并将undo和redo信息写入
日志进行持久化。
所有参与者响应协调者发起的询问。对于每个参与者节点，如果他的事务操作执行成
功，则返回“同意”消息；反之，返回“终止”消息。
commit阶段
如果协调者获取到的所有参与者节点返回的消息都为“同意”时，协调者向所有参与者节
点发送“正式提交”的请求（成功情况）；反之，如果任意一个参与者节点预提交阶段返
回的响应消息为“终止”，或者协调者询问阶段超时，导致没有收到所有的参与者节点的
响应，那么，协调者向所有参与者节点发送“回滚提交”的请求（失败情况）。
成功情况所有参与者节点正式完成操作，并释放在整个事务期间占用的资源；反之，失
败情况下，所有参与者节点利用之前持久化的预写日志进行事务回滚操作，并释放在整
个事务期间占用的资源。
成功情况下，所有参与者节点向协调者节点发送“事务完成”消息；失败情况下，所有参
与者节点向协调者节点发送“回滚完成”消息。
成功情况下，协调者收到所有参与者节点反馈的“事务完成”消息，完成事务；失败情况
下，协调者收到所有参与者节点反馈的“回滚完成”消息，取消事务。

如果你看到一个 task 发生 反压警告（例如： High ），意味着它生产数据的速率比下游 task 消
费数据的速率要快。
在工作流中数据记录是从上游向下游流动的（例如：从 Source 到 Sink）。反压沿着相反的方向传
播，沿着数据流向上游传播。

反压是流处理系统中用来保障应用可靠性的一个重要机制。由于流应用是7*24小时运行，数据输
入速率也不是一成不变，可能随时间产生波峰波谷，当某个处理单元由于到来的数据忽然增加，暂
时性超出其处理能力时，就会出现数据在接收队列上累积，当数据的累积量超出处理单元的容量
时，会出现数据丢失现象甚至因为系统资源耗尽而导致应用崩溃。为此，需要一种反压机制来告知
上游处理单元降低数据发送的速率，以缓解下游处理单元的压力。
流处理系统需要能优雅地处理反压（backpressure）问题。反压通常产生于这样的场景：短时负
载高峰导致系统接收数据的速率远高于它处理数据的速率
反压并不会直接影响作业的可用性，它表明作业处于亚健康的状态，有潜在的性能瓶颈并可能导致
更大的数据处理延迟。
通常来说，对于一些对延迟要求不太高或者数据量比较小的应用来说，反压的影响可能并不明显，
然而对于规模比较大的 Flink 作业来说反压可能会导致严重的问题。
反压会影响到两项指标: checkpoint 时长和 state 大小。
前者是因为 checkpoint barrier 是不会越过普通数据的，数据处理被阻塞也会导致
checkpoint barrier 流经整个数据管道的时长变长，因而 checkpoint 总体时间（End to End
Duration）变长。
后者是因为为保证 EOS（Exactly-Once-Semantics，准确一次），对于有两个以上输入管道
的 Operator，checkpoint barrier 需要对齐（Alignment），接受到较快的输入管道的
barrier 后，它后面数据会被缓存起来但不处理，直到较慢的输入管道的 barrier 也到达，这
些被缓存的数据会被放到state 里面，导致 checkpoint 变大。

对于一个 TaskManager 来说会有一个统一的 Network BufferPool 被所有的 Task 共享，在初始化
时会从 Off-heap Memory 中申请内存，申请到内存的后续内存管理就是同步 Network
BufferPool 来进行的，不需要依赖 JVM GC 的机制去释放。有了 Network BufferPool 之后可以为
每一个 ResultSubPartition 创建 Local BufferPool 。
如上图左边的 TaskManager 的 Record Writer 写了 <1，2> 这个两个数据进来，因为
ResultSubPartition 初始化的时候为空，没有 Buffer 用来接收，就会向 Local BufferPool 申请内
存，这时 Local BufferPool 也没有足够的内存于是将请求转到 Network BufferPool，最终将申请
到的 Buffer 按原链路返还给 ResultSubPartition，<1，2> 这个两个数据就可以被写入了。之后
会将 ResultSubPartition 的 Buffer 拷贝到 Netty 的 Buffer 当中最终拷贝到 Socket 的 Buffer 将
消息发送出去。然后接收端按照类似的机制去处理将消息消费掉。
因为速度不匹配就会导致一段时间后 InputChannel 的 Buffer 被用尽，于是他会向 Local
BufferPool 申请新的 Buffer ，这时候可以看到 Local BufferPool 中的一个 Buffer 就会被标记为
Used
发送端还在持续以不匹配的速度发送数据，然后就会导致 InputChannel 向 Local BufferPool 申请
Buffer 的时候发现没有可用的 Buffer 了，这时候就只能向 Network BufferPool 去申请，当然每
个 Local BufferPool 都有最大的可用的 Buffer，防止一个 Local BufferPool 把 Network
BufferPool 耗尽。这时候看到 Network BufferPool 还是有可用的 Buffer 可以向其申请。
显然，再过不久 Socket 的 Buffer 也被用尽，这时就会将 Window = 0 发送给发送端（前文提到的
TCP 滑动窗口的机制）。这时发送端的 Socket 就会停止发送。
很快发送端的 Socket 的 Buffer 也被用尽，Netty 检测到 Socket 无法写了之后就会停止向 Socket
写数据。
Netty 停止写了之后，所有的数据就会阻塞在 Netty 的 Buffer 当中了，但是 Netty 的 Buffer 是无
界的，可以通过 Netty 的水位机制中的 high watermark 控制他的上界。当超过了 high
watermark，Netty 就会将其 channel 置为不可写，ResultSubPartition 在写之前都会检测
Netty 是否可写，发现不可写就会停止向 Netty 写数据。
这时候所有的压力都来到了 ResultSubPartition，和接收端一样他会不断的向 Local BufferPool
和 Network BufferPool 申请内存。
Local BufferPool 和 Network BufferPool 都用尽后整个 Operator 就会停止写数据，达到跨
TaskManager 的反压。

下游的 TaskManager 反压导致本 TaskManager 的 ResultSubPartition 无法继续写入数据，于是 
Record Writer 的写也被阻塞住了，因为 Operator 需要有输入才能有计算后的输出，输入跟输出
都是在同一线程执行， Record Writer 阻塞了，Record Reader 也停止从 InputChannel 读数据，
这时上游的 TaskManager 还在不断地发送数据，最终将这个 TaskManager 的 Buffer 耗尽。