1. 概念

 

有状态的流式计算框架

可以处理源源不断的实时数据，数据以event为单位，就是一条数据。

2. 开发流程

先获取执行环境env，然后添加source数据源，转换成datastream，然后使用各种算子进行计算，使用sink算子指定输出的目的地，最后调用execute方法执行。

3. flink运行模式

1. standalone
2. yarn
3. k8s

4. flink部署模式(yarn)

1. session
   1. 先启动集群，再提交job到集群
2. per-job
   1. 一个job启动一个集群
3. aplication
   1. 一个job启动一个集群

per-job和application区别：

• 提交代码位置不一样，单作业模式的main方法在客户端执行，应用模式的main方法在JobManager执行

应用模式是生产上主要提交模式，单作业模式和应用模式都是一个job启动一个集群，所以可以做到资源隔离，而会话模式是多个job分享一个集群，适合小作业共享。

5. 运行时架构

1. Client
   1. 解析代码，提交作业

2. JobManager

   1. 管理节点，任务切分分配

   2. dispatcher：将job传递给Jobmaster

   3. resourManager：申请资源

   4. JobMaster：切分任务

   5. Checkpointcoordinator：向数据源注入barrier

3. TaskManager

   1. 执行任务计算 

   2. 资源最小单位slot ，算子就是我们task任务

6. 基本概念

6.1.task和subtask区别？

一个算子（map，filter，flatmap）就是一个task

算子并行子任务就是subtask 

6.2 task和slot的关系

一个task的子任务不能在一个slot中执行

一个slot中可以执行不同算子的subtask

6.3 并行度的优先级

算子 ＞ 全局env  ＞ 提交命令行  ＞   配置文件

6.4 算子链路的合并

多个subtask组成一个大的subtask

条件：

1. 前后算子的并行度一致
2. forward（数据分区规则）
3. subtask必须在一个共享槽（.slotSharingGroup("default"), 在一个slot槽中执行）

算子合并优点和缺点 ？

1. 优点
   1. 节省数据传输IO
2. 缺点
   1. 如果有subtask计算逻辑复杂会有抢占资源问题

如何禁用算子链？

env.disableOperatorChaining()

如何设置不同的共享槽？

.slotSharingGroup("aa")

6.5 流图转化

	产生	发送	做了什么事情
StreamGraph	Client	Client	代码解析
JobGraph	Client	JM	算子链的合并
ExecutionGraph	JM	TM	并行子任务显示
物理执行图

6.6 per-job模式提交作业流程

1. 客户端提交代码，解析参数 生成StreamGraph
2. 由StreamGraph生成jobGraph，主要是做了算子链合并
3. 封装参数 提交给集群yarn 的RM
4. yarn找一个NM，启动JM
5. 启动dispatcher，RM，Jobmaster，生成executionGraph
6. 向JM的RM申请资源，然后去找Yarn的RM申请资源，创建TM启动slot 
7. 注册slot，分配任务

7. API

7.1 source

kafkasource（算子状态，保存offset） 

7.2 transform

1. 单流：map，flatmap，filter
2. keyby ：sum， min， max ，reduce 
3. 侧输出流
4. 物理分流算子：shuffle，forwawrd，rebalance（默认），rescale
5. union（类型要求一致）  connect（可以不一致）

7.3 sink

kafkasink，dorissink， jdbcsink， filesink 

7.4 join

1. API
   1. windowjoin
   2. interval join ：两条实时流去根据范围关联，如果一些迟到特别久的数据关联不上
2. SQL
   1. 常规join（比如left join ，支持回撤流）
   2. lookupjoin：读取外部系统数据，可以缓存， 适用于数据量小，而且基本不变化的表（比如字典表）
   3. interval join
   4. window tvf函数 ：累积函数，滚动，滑动

8. 时间语义

1. 事件时间：业务数据推动，获取数据中时间戳，推进时间
2. 处理时间：获取操作系统时间
3. 摄入时间：数据进入到flink集群的系统时间

• 共同点
  • 时间不能倒退，单调递增 
• 区分
  • （处理时间）速度稳定，不能停滞 
  • （事件时间）速度不稳定，可能会停滞

9. WaterMark

9.1 你对watermakr的理解

逻辑时钟，单调递增，解决乱序迟到问题

9.2 水位线传递

• 一对多：广播水位线
• 多对一：取最小
• 多对多：先广播，再取最小

场景题：上游算子发生数据倾斜，某一个subtask没有数据，水位线无法抬升怎么办？

解决办法：
            调用withIdleness()方法，如果某一个subtask没有数据，超过了空闲等待时间，那么放弃使用这个subtask的水位线。

9.3 迟到数据问题如何解决？

1. 设置乱序时间：针对于迟到时间短的数据
2. 窗口延迟关闭：迟到中级
3. 侧输出流：迟到特别长

9.4 水位线注入规则

当前最大时间戳 - 乱序时间  - 1ms

10. 窗口

概念：无界流切分为有界流， 集合中是一个个的桶

10.1 分类

1. 滑动
2. 滚动
3. 会话：按照时间间隔划分窗口

10.2 四大组成

1. assigner：分配器
2. trigger ：触发窗口计算
3. evictor：驱逐器，清除窗口数据
4. 聚合逻辑：增量聚合， 全量聚合(reduce    aggregate)

场景问题：表的字段有mid  timestamp  price   ，要求算当前累积GMV， 5分钟输出一次

解决方案：

1. 第1种方案：windowtvf函数 Cumulate Windows
2. 第2种方案：用滚动窗口 1天  ，实现ContinuousEventTimeTrigger，自定义每5分钟输出一次

10.3 核心概念

划分（数据属于哪个窗口）

开一个5s滚动窗口  数据是3s  会落到哪个窗口：0-5   3-8 

结论：窗口的向下取整
                timestamp - (timestamp - offset) % windowSize

生命周期

创建：属于窗口第一条数据到来

销毁：事件时间 ＞＝ 窗口长度 +允许迟到时间

左闭右开

        endtime -1ms

10.4 设置乱序时间 和窗口延迟关闭时间 有什么区别？

5s滚动窗口   乱序时间设置2s   销毁时间5s  （7s数据过来时候，时间推进到5s）

5s滚动窗口   窗口延迟关闭2s   销毁时间7s   （7s数据过来时候，时间推进到7s）

结论：

设置乱序时间，并不会影响窗口销毁时间，影响时间推进规则，窗口延迟关闭时间影响窗口的关闭时间。

举个栗子：

10s滚动窗口，设置乱序时间5s，窗口延迟关闭时间5s 

窗口销毁：水位线15s时候销毁， 数据携带20s及以上过来触发窗口销毁

11. 状态

概念：用户定义的一些变量 

状态数据是交由Flink托管的，考虑程序数据的恢复 

11.1 分类

1. 算子状态：每个subtask
   1. list：恢复状态时候轮询
   2. unionlist：广播
2. 键控状态：每个key去维护的状态
   1. value  map  list  reduce  aggregate 

11.2 状态后端

	本地	远端
hashmap	TM堆内存	hdfs
rocksdb	rocksdb	hdfs

使用场景：rocksdb存储数据量级别比hashmap大

11.3 状态后端场景选择

企业中大状态场景选用的rocksdb  ，大状态场景优化

举个例子：

用户新老访客修复  1000w用户    1k      ≈  10G

rocksdb支持：增量检查点  、 本地恢复 、预定义选项

11.4 TTL

状态的过期时间是由哪个类设置的：

StateTttlConfig

12. 容错机制

12.1 端到端一致性 （kafka  flink   kafka）

源头：offset可重发

Flink：checkpoint

sink：事务（2pc 预写日志） 幂等 

12.2 checkpoint流程

1. JM的checkpoint协调器发送命令startcheckpint开始
2. 定期向数据源注入barrier （特殊事件，不会跳过数据向下游发送）
3. barrier随数据流过每个subtask 
4. barrier到每个算子，将本地状态快照到hdfs文件系统，快照完之后acks应答（barrier之前的数据已经进入kafka，预提交）
5. JM中协调器收到所有算子的acks，标志所有快照做完，向算子分发消息
6. 正式提交kafka

12.3 barrier

1. 精确一次性
   1. barrier对齐：等待所有barrier到来，快照，等待的时候将数据缓存不处理
   2. 1.11版本，barrier不对齐，状态数据和缓存数据同时快照
2. 至少一次
   1. barrier对齐：等待所有barrier到来，快照，数据直接向下游传递，不阻塞在缓存中
   2. 问题：出现意外恢复，状态中有重复数据问题

12.4 savepoint 和checkpoint区别

1. checkpoint：自动帮我做
2. savepoint手动：配置文件指定savepoint的路径，取消任务触发保存点停止

场景：程序升级 （算子增加，算子减少）
            增加uid

13. FlinkSql

Flinksql如何转化成底层的api？

使用calcite解析语法树

sql转化  ast语法树   逻辑执行    物理执行     底层api执行 

14. Flink生产经验

14.1 提交任务脚本

bin/flink run 
        -d   后台运行 
        -D   并行度     5
        -D   JM内存     1~4 G   
        -D   TM内存     4~8 G 
        -D   TM的slot个数  3（1~4）  
        -c  主类
        ./jar包
        

如果并行设置为5个，slot个数设置为3个，那么会启动2个TM

14.2 TM内存模型

1. JVM
   1. 元空间
   2. 执行开销
2. FLink内存
   1. 堆内：框架内存，task计算内存（分配，剩余内存）
   2. 堆外：框架内存，task计算内存（0）  网络内存（组件之间交互，算子缓存区）  托管内存（状态数据）

14.3 Flink部署多少台机器

FLink充当客户端， ds的worker节点都需要部署 

如果是streampark：需要部署一台

15. Flink和sparkstreaming区别 /Flink优点

	Flink	sparkstreaming
模型	流式	微批次
时间	丰富	处理时间
乱序	解决	不能解决
窗口	多灵活	窗口长度必须是批次整数倍
容错机制	有	没有
状态	有	没有

16. Flink的Interval Join的实现原理？Join不上的怎么办？

底层调用的是keyby + connect ，处理逻辑：

（1）判断是否迟到（迟到就不处理了，直接return）

（2）每条流都存了一个Map类型的状态（key是时间戳，value是List存数据）

（3）任一条流，来了一条数据，遍历对方的map状态，能匹配上就发往join方法

（4）使用定时器，超过有效时间范围，会删除对应Map中的数据（不是clear，是remove）

Interval join不会处理join不上的数据，如果需要没join上的数据，可以用 coGroup+join算子实现，或者直接使用flinksql里的left join或right join语法。

17. Flink的keyby怎么实现的分区？分区、分组的区别是什么？

分组和分区在 Flink 中具有不同的含义和作用：

分区：分区（Partitioning）是将数据流划分为多个子集，这些子集可以在不同的任务实例上进行处理，以实现数据的并行处理。

数据具体去往哪个分区，是通过指定的 key 值先进行一次 hash 再进行一次 murmurHash，通过上述计算得到的值再与并行度进行相应的计算得到。

分组：分组（Grouping）是将具有相同键值的数据元素归类到一起，以便进行后续操作（如聚合、窗口计算等）。key 值相同的数据将进入同一个分组中。

注意：数据如果具有相同的 key 将一定去往同一个分组和分区，但是同一分区中的数据不一定属于同一组。