Flink总结

从头儿过一遍书，做了些摘要。SQL那里还没仔细复习。

一、初始Flink

1. 核心目标：数据流上的有状态计算

2. 具体定位：以内存执行速度（速度快）和任意规模来执行计算（可扩展性强） -> 小松鼠快速灵巧

3. 有状态的流处理可用于许多不同场景：

   • 事件驱动型应用：以Kafka为代表的消息队列几乎都是事件驱动型应用。因为有状态，不再需要查询数据库，而是本地访问数据。这样在吞吐量和延迟上可以有更好的性能。
     • 检查点可以异步增量地完成，对正常计算影响非常小。
   • 实时数据分析：利用状态
   • 数据管道：转换或扩展数据，在存储系统之间移动数据。

特点：

1. 高吞吐和低延迟。
2. 结果准确，乱序事件流，事件时间语义仍可以提供一致且准确的结果。
3. 精确一次的状态一致性。
4. 高可用。
5. 可以连接到常用存储系统。

Spark vs Flink

• 低延迟流处理选Flink，因为Spark要攒批，无法在低延迟做到极致。
• 海量数据批处理选Spark，吞吐量更大，生态更完善，API成熟易用。

比较：

1. 数据模型不同。
2. 计算逻辑：Spark做批计算，需要将任务对应的DAG划分阶段，一个完成后经过shuffle再进行下一阶段的计算。Flink是标准的流式执行模式，一个事件在一个节点处理完后可以直接发往下一个节点进行处理。
3. Flink优势：
   • 毫秒级延迟
   • 严格精确一次语义
   • 窗口API更灵活，语义更丰富
   • 提供事件时间语义，可以正确处理延迟数据。
   • 提供了状态。

三、Flink部署

web UI : 8081

Standalone模式、K8s模式略过。重点写YARN模式

部署模式分类

主要区别在于：集群的生命周期、资源的分配方式、应用的main方法到底在哪里执行（客户端还是JM）

1. 会话模式(Session Mode)
   • 先启动一个集群，保持一个会话，在这个会话中通过客户端提交作业。
   • 因为资源共享，资源不够时新作业失败。
   • 同一个TM可能运行了很多作业，其中一个故障导致TM宕机 -> 影响其他作业
2. 单作业模式(Per-Job Mode)
   • 每个提交的作业启动一个集群。（更好地隔离资源）
   • 客户端运行app后，启动集群；作业完成后，集群关闭。
   • 更加稳定，Per-Job模式需要借助资源管理框架来启动集群。
3. 应用模式(Application Mode)
   • 前面两种方法应用代码都是在客户端执行，然后由客户端提交给JM。
   • 客户端需要占用大量网络带宽，将数据发送给JM。会加重客户端节点的资源消耗。
   • 不要客户端了，直接把app提交给JM。-> 为每个提交的任务单独启动一个JM(创建一个集群)
   • 这个JM只为执行这一个app而存在，执行结束后JM关闭。

YARN模式

客户端把Flink应用提交给YARN的ResourceManager，RM向NodeManager申请容器。在容器上，Flink会部署JobManager 和 TaskManager的实例，从而启动集群。

Flink会根据运行在JM上的作业所需要的Slot数量动态分配TM资源。

高可用

YARN的高可用，只启动一个JM，当这个JM挂掉之后，YARN会再次启动一个，利用YARN的重试次数来实现的高可用。

四、Flink运行时架构

JM

任务管理和调度的核心

1. JM接收客户端提交的 Jar包、数据流图(dataflow graph)、作业图(Job Graph)
2. JM将作业图转换成物理层面的数据流图 -> 执行图(Execution Graph) 包含了所有可以并发执行的任务
3. JM向RM申请资源，申请到后，将执行图发到TM上。
4. RM的资源就是slot资源，包含一组执行计算的CPU和内存资源。
5. 分发器Dispatcher，用来提交应用。

TM

工作进程

1. 每一个TM都包含了一定数量的任务槽。slot是资源调度的最小单位，slot的数量限制了TM能够并行处理的任务数量。
2. 启动TM后，TM会向RM注册slots，收到RM指令后，TM会将一个或者多个slot提供给JM调用。

宏观流程

1. 客户端通过分发器，将作业交给JM
2. 分发器启动JM，将作业和作业图提交给JM
3. JM将作业图解析为可执行的枝形图，得到资源数量，向RM请求slots
4. RM判断当前资源是否足够，不够则启动新的TM
5. TM启动后，向RM注册slots
6. TM连接到JM，提供slots
7. JM将所需任务分发给TM
8. TM执行任务

YARN-Per-Job流程

1. 客户端将作业提交给YARN的RM，会将jar包和配置上传到HDFS
2. RM启动容器资源，启动JM，并将作业提交给JM
3. JM向RM(f)请求slots资源
4. RM(f)向RM请求容器资源
5. YARN启动新的TM容器
6. TM启动之后，向RM(f)注册可用的slots
7. RM(f)通知TM为新作业提供slots
8. TM连接到JM，提供slots，JM将任务分发给TM执行任务。

重要概念

并行度

前一个操作处理完成，就发往下一步操作的节点。

任务有先后，这里关心的是数据的并行。

每一个算子可以包含一个或多个子任务，这些子任务在不同的线程、不同的物理机或不同的容器中完全独立地执行。

一个特定算子的子任务个数，被称为它的并行度。一个流程序的并行度，可以认为是其所有算子中最大的并行度。

并行度设置：代码中指定 > 提交时的命令行参数 > 配置文件

算子链

并行度相同的one to one 算子操作，可以直接链接在一起形成算子链。可以减少线程之间的切换和基于缓冲区的数据交换，减少时延的同时提升吞吐量。

四张图

逻辑流图 -> 作业图 -> 执行图 -> 物理图

任务和任务槽

1. slots
   • 一个TM是一个JVM进程。可启动多个独立的线程，并行执行多个子任务。
   • 为了控制并发量，在TM上对每个任务所占用的资源做出明确划分 -> slot
2. slots数量
   • 如果只要一个slot，那么每个任务运行在独立的JVM中。
   • 如果有多个JVM，则多个任务共享一个JVM。
   • slot只隔离内存，不隔离CPU
3. 任务对slot的共享
   • 同一个作业，不同任务节点的并行子任务，可以放在同一个slot执行。

五、DataStream API

数据类型

Flink的类型基类 TypeInformation来同一表示数据类型。

由于Java泛型擦除的存在，在某些特殊情况（比如Lambda表达式），自动提取的信息推断不出类型。需要显示地提供类型信息。(.returns(Types.TUPLE(Type.STRING, Types.LONG)))

算子分类

1. 转换算子
   • map, filter, flatMap
2. 聚合算子
   • keyBy(聚合前要分区)
   • sum
   • min
   • max
   • minBy ，min只计算指定字段的最小值，minBy会返回包含字段最小值的整条数据
   • maxBy
   • reduce 一般化聚合
3. UDF

4. 富函数类

   • 所有的Flink函数都有其Rich版本，一般以抽象类形式出现。
   • 复函数可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂功能。
   • open()方法，初始化方法，文件IO创建、数据库连接、配置文件读取等一次性工作可以放在open中
   • close()，最后一个调用的方法啊，可以做清理用。
   • 富函数类提供了 getRuntimeContext()方法，可以获取运行时上下文信息，比如程序执行的并行度、任务名、状态.

5. 物理分区

   • keyBy是逻辑分区，可能发生数据倾斜，或者需要手动分配分区策略时，可以使用物理分区。
   • 随机分配 .shuffle() 将数据按均匀分布随机传递到下游，每次执行结果不同。
   • 轮询分配 .rebalance()实现轮询重分区
   • 重缩放 .rescale() 也是轮询，只是当多个上游和多个下游时，轮询时1对all，冲缩放是1对n，多个上游共同分配all。
   • 广播 .broadcast() 数据会在不同分区都保留一份儿，可能进行重复处理。将输入数据复制并发送到下游算子的所有并行任务中去。
   • 全局分区 .global() 将所有的输入流数据都发送到下游算子的第一个并行子任务中去。强行让下游任务并行度变为1.
   • 自定义分区 .partitionCustom()

Sink算子

状态为啥不写入redis呢？因为写入redis一旦发生故障，需要复杂的机制保证恢复到之前的状态。

Flink内部提供了checkpoint来保证可以回滚到正确的状态。但是如果在处理过程中任意读写到外部系统，发生故障后就很难回退到从前了。

Flink与Kafka的连接器提供了端到端到精确一次语义。

• 自定义sinkFunction
  • 比如自定义到HBase的函数
  • 使用RickSinkFunction，open连接，close关闭连接
  • 在invoke方法中实现插入逻辑

六、Flink中的时间和窗口

在流式处理的过程中，数据是在不同的节点间不停流动的，由于网络延迟，上下游任务对于时间的理解也会有所不同。

时间语义

• 事件时间
  • 每个事件在对应的设备上发生的时间，也就是数据生成的时间
  • 与机器无关，只依赖于数据本身
• 处理时间
  • 不考虑节点之间的协调同步，不需要考虑数据在流中的位置
• 摄入时间
  • 数据进入Flink淑女刘的时间，也就是Source算子读入数据的时间
  • 是对处理时间和事件时间的中和。

• 时间语义无绝对好坏
  • 处理时间用于实时性要求极高，但是计算准确性要求不太高的场景
  • 事件时间更符合业务场景，在事件时间语义下，水位线成了时钟，可以同一控制时间的进度，保证了我们总可以将数据划分到正确的窗口中。

水位线

用来度量事件时间，水位线之前的数据全部到达了。

划分了窗口后，根据数据的时间戳确定其属于哪个窗口。窗口处理的是有界数据，需要等窗口数据全部到齐才能计算出最终的统计结果。

本身窗口的时间进度可以通过时间戳标识，但是上下游之间的时间同步怎么办呢？比如下游有三个并行度，其中一个窗口接收到了9点的数据，窗口开始聚合运算，但是其他两个子任务并没有接收到这个数据，不能进行聚合运算。

水位线就是在数据流中加入一个时间标记，记录当前的事件时间，这个标记可以广播到下游，当下游任务收到这个标记，就可以更新自己的时钟。

• 有序流的WM：周期插入（周期时间是事件时间）
• 乱序流的WM：时间戳有更新时，才插入WM。大量数据到来时，WM插入会影响效率，这时可以周期性插入WM，保存当前的最大值即可，插入时插入最大的。
  • 处理迟到，等一段时间，即WM作为时钟，把时间拨慢一点儿。
    • 在注册水位线的时候，给定延迟
    • 时间戳是当前最大时间-1ms

窗口

Flink中，窗口

设置2s的延迟，窗口是左闭右开的，当12s的数据到来，[0,10)窗口关闭，12s数据进入到[10,12)中。

时间窗口、计数窗口：

• 滚动窗口：对时间段做聚合统计，可以应用于很多BI分析指标
• 滑动窗口：滑动步长小于窗口大小，并尽量设置为整数倍关系。
  • 统计最近一段时间的指标，结果的输出频率要求很高，甚至要求实时更新。
  • 或者基于一段时间行为检测的异常报警。
• 会话窗口：只能基于时间定义，没有会话计数窗口。
  • gap时间内没有数据，就关闭窗口。
  • 每新来一个数据，都会创建一个会话窗口，判断和已有窗口的距离，小于size就合并窗口。
• 全局窗口：默认不触发计算，需要自己定义触发器。

是否按key分区

窗口函数 Window Function

• 增量聚合
  • 来一条计算一条，窗口结束再输出。
  • Reduce Function ： 聚合状态的类型、输出结果的类型都必须和输入类型一样
  • Aggregate Function ： 可以灵活定义<IN, ACC, OUT>
• 全窗口聚合
  • 先收集并内部缓存，等到窗口要输出结果的时候再取出数据计算。
  • .apply() 传入一个Window Function实现类。可以得到窗口信息
  • .process() 传入Process Window Function实现类，可以获得上下文对象 -> 可以访问时间和状态。
• 增量聚合更高效，全窗口信息更多。
• 在调用增量聚合时，可以再传入一个全窗口函数。
• 

迟到数据处理

1. 水位线，拨慢时钟

   

2. 允许延迟，在不考虑水位线的情况下，使窗口晚一点再销毁。

3. 侧输出流接收迟到数据。

怎么将侧输出流的数据同步到之前的结果中呢？

七、处理函数

处理函数中，直面的就是数据流中最基本的元素：数据事件Event、状态State、时间time

富函数类：可以拿到状态、并行度、任务名等运行时信息

top N

• 窗口聚合后，将窗口的end时间放入POJO
• 之后根据end key by
• 然后将同一end的数据，存入状态，取top N

分流

可以利用处理函数分流，利用侧输出流分流。

八、多流转换

分流

1. 侧输出流（推荐）
2. filter 将一条流重放，不现实
3. split 方法已经弃用

基本合流

1. 联合 （Union）
   • 流中数据类型必须相同，暴力交汇
   • 不同流中水位线进度不同，合流之后的水位线以最小的为准。
2. 连接 （Connect）
   • 允许类型不同，得到的是连接流(Connected Streams)。形式“统一”，彼此之间独立。
   • 变回Data Stream 定义co-process 说明对于不同的流分别怎么处理。
   • 可调用的co-process方法有 map、flatMap、process，传入一个Co***Function实现类
3. 广播连接流 （BroadcastConnectedStream）

基于时间的合流--双流join

两条流的合并，前面都是将流放在一起，我们有时更希望根据某个字段的值将流联结起来，配对去做处理。

也可以利用Connect和key by实现，不过有些抽象。

窗口联结 Window Join

间隔联结 Interval Join

窗口同组联结 Window CoGroup

双流join优化与总结

1. 双流join时间到了不触发、没输出
   • wm是否合理，数据事件是否远远大于wm和窗口时间
2. 双流join利用的是state数据，state保存多久，会内存爆炸吗？
   • state自带有ttl机制，可以设置ttl过期策略，建议程序中的state用完之后手动clear
3. 双流join数据倾斜
   • 过滤异常key
   • 拆分表减少数据
   • 打散key分布
4. 多流join
   • 先union再后续处理
   • 先connect再join
5. join过程延迟，没有关联上的数据会丢失吗？
   • 侧输出流可以存储延迟流
   • 节点网络异常，检查点可以保证数据不丢失。

九、状态编程

状态由任务维护，用来计算输出结果的所有数据。

有状态算子：聚合算子、窗口算子

无状态算子：基本转换算子

状态

• 算子状态：范围限定为当前的算子任务实例，支队当前并行子任务有效。
• 按键分区状态：再只对当前key有效

按键分区状态 Keyed State

可以通过复函数获得状态

• 值状态 ValueState
• 列表状态 list
• 映射状态 map
• 归约状态 Reducing State
  • 类似于值状态，需要对添加进的所有数据进行归约
• 聚合状态 Aggregating State
  • 类似于归约状态，传入的函数更加一般化

状态生存时间 TTL

算子状态

应用场景较key state少，主要应用在Source 和 Sink到外部系统的算子上，或者完全没有key的场景。

Flink的Kafka连接器使用了算子状态。Kafka消费者的每一个并行实例，都会为对应的主题(topic)分区维护一个偏移量，作为算子状态保存起来。

广播状态

状态后端

状态后端分类：

• 哈希表状态后端：本地状态存入内存（把状态当做对象，保存在TM的JVM堆上
• RocksDB状态后端：本地状态存入RocksDB

十、容错机制

checkpoint

在有状态的流处理中，任务继续处理新数据，并不需要“之前的计算结果”，而是需要任务“之前的状态”。实现容错 -> 将某个时间点的状态保存下来，这份存档即是 checkpoint

周期性存档

状态的恢复：精确一次的状态一致性保证，故障发生前的结果并没有保存到新状态中，不会对结果产生影响。

检查点算法：分布式快照

状态一致性

• 最多一次
  • 只求快，不求准的场景
• 至少一次
  • 算具有幂等性的场景时可以使用，比如计算UV
  • 需要在发生故障时可以重放数据。
• 精确一次
  • 

端到端精确一次

• source端

  • 数据源有重放数据的能力。
    • 对数据进行持久化保存，并且可以重设数据的读取位置。
  • Flink的Source任务中将读取的偏移量保存为状态，这样就可以在故障恢复时从检查点读取出来，对数据源重置偏移量，重新获取数据。

• flink内部：检查点机制，在能够重放数据流的情况下，可以保证精确一次。

  • 检查点能保证精确一次，不是之前处理过，恢复过后就不处理了。
  • 而是呀看状态的改变和输出的结果，是否只包含了一次这个数据的处理。

• sink端

  • 数据有可能重复写入外部系统
  • 
    • 幂等写入：多次写入，但是只产生一次结果。类似于HashMap，但是并没有真正解决重复计算写入问题。
    • 
    • 事务写入：利用事务解决外部系统写入无法撤销的问题。
    • 
      • 预写日志：WAL
      • 
      • 
      • 二阶段提交：2PC
      • 
      • 

Flink和Kafka连接时的精确一次保证

• Source
  • kafka可以对数据进行持久化保存，可以重置偏移量 offset
  • FlinkKafkaConsumer可以将当前读取的偏移量保存为算子状态，写入检查点中。
  • 当发生故障时，从检查点读取恢复状态，并由连接器FlinkKafkaConsumer向Kafka重新提交偏移量，就可以重新消费数据、保证结果的一致性了。
• Flink内部
  • 检查点机制
• Sink
  • 采用2PC方式，处理完毕得到结果，写入Kafka是基于事务的预提交
  • 等到检查点保存完毕，才会提交事务进行正式提交
  • 如果出现故障，事务回滚，预提交就会被放弃
  • 恢复状态之后，也只能恢复所有已经确认提交的操作。

需要的配置

十一、Table API和SQL

大致流程：

• 流执行环境
• 流读取数据
• 表执行环境
• 流转表
• 处理逻辑
• 表转流

流处理中的表

十二、Flink CEP

复杂时间处理 Complex Event Processing

• 定义匹配规则
• 将模式应用到流上
• 检测到复杂时间进行处理，得到结果输出

参考

《剑指大数据》