Flink1.14知识点总结

0 初识Flink

1. 批计算vs流式计算?

   明确一点,对比两种计算方式本质就是区别两种流(有界流和无界流);批计算输入的是有界流即处理前为完成的数据集,因此输出得到的就是最终确定的报表;流式计算输入的是无界流,输入永不结束,由于任一时刻的处理前数据并不是一个整体,因此其输出类似地也并不是最终结果,同样是无界的结果流

2. Flink vs Spark?

   ①分布式计算:spark有上下游间的shuffle以及Executor向Driver汇总,flink有多任务实例

   ②流/批计算:flink流批一体,spark批处理有sparkcore和sparksql,流式处理有sparkstreaming

   ③flink自带状态管理机制,spark没有

   ④job执行:spark由sql到树到执行计划,flink由代码到图到执行计划

   ⑤flink每一个算子都可扩展,spark不行,阶段划分(shuffle)后多个算子合并才可扩展,当然flink有合并算子链类似该机制,因此flink可更灵活控制上下游算子并行度变化,而spark无法实现(理解flink算子粒度级独立并行度)

3. Flink基本介绍?

   流式计算为基础,引入了有界流的批计算,从而实现了流批一体

4. Flink核心架构?

   主从架构:JobManager为Master,TaskManager为Slave,以及Client

   各角色职责:不同的部署模式下各角色职责会有变动,想要理解各角色的职责,其实只需要从执行job的角度看各组件做了什么事就能明白

   job执行流程(以standalone集群为例):job提交->客户端调用用户主类main方法,解析生成流图,加工成作业图,发送给JobManager->JobManager将其加工得到执行图,并将图中各任务在TaskManager的Slot上执行

5. 简述Flink的优势?

   流批一体、事件驱动(事件时间语义+ETL)、自带状态管理机制、容错(状态一致性+事件时间语义+迟到数据处理)、多层api、分布式架构等

6. 初步多角度理解flink状态?

   ①状态是什么? 一切皆状态,flink中任何地方都可以有状态,除去用户自行调用api获取状态,状态随处可见,像窗口的桶中存放的就是、聚合算子的临时结果(累加器)也是,一旦你需要对输入的数据或者临时计算结果攒批时,难免用到状态

   ②状态放在哪? 运行时的状态由状态后端管理,存放的原则为少则放于内存,多则溢出到磁盘,因此不用担心oom

   ③状态后端如何管理状态? 运行时状态以对象/字节的形式保存,而快照时则写出到外部存储,还负责状态恢复和过期清理

1 运行时环境

1. 简单理解flink的数据抽象DataStream?

   首先,数据流DataStream可以是有界也可以是无界,因为api层面已经实现流批一体;其次DataStream不可变,flink之所以能对数据进行转换,和spark的rdd类似,并不是直接改变rdd/流,而是生成新的rdd/流从而调用算子中逻辑

2. 流批一体在api中的体现?

   //老版本flink区别流/批计算在于入口不同
   
   //批处理计算入口
   ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
   
   //流式处理计算入口
   StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
   
   //新本版flink已实现流批一体入口
   StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
   
   //上面入口默认为流处理模式,可按需手动设置为批处理模式
   ExecutionConfig config = env.getConfig();
   config.setExecutionMode(ExecutionMode.BATCH);
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14

3. 既然已实现流批一体的api,干嘛还提供可将配置改为批处理模式?

   首先看流,无界流自动不用改配置,若是有界流则流批一体入口默认流计算模式,虽不会报错但是与批计算模式底层分布式计算原理却有差异,流计算来一条处理一条批计算攒批处理,所以对于有界流批计算的shuffle中的优化机制更有优势,因此需手动设置为批处理模式,批计算中,数据抽象是DataSet数据集

   生产中,正常跑用流式计算,重跑数据、补数,批处理包含shuffle的预聚合等优化,效率更高,因此通过提交命令时配置批处理模式参数即可

2 基本算子介绍

1. 基于文件的source算子?

   readFile读取文件数据,传入文本解析格式、路径、处理模式,处理模式为PROCESS_ONCE时文件读一次就推出了,PROCESS_CONTINUOUSLY时会一直监听文件,一旦文件内容变化则整体重读因此会造成数据重复读取,而readTextFile算子底层为传参固定的readFile

2. 如何理解Kafka Source细节?

   ①kafka回顾:消费者offset间隔固定时间自动提交,调用consumer.commitAsnyc()提交到_consumers_offset主题,生产者端支持事务,消费者端不支持事务但消费者的更新偏移量操作一般和生产者的写数据操作事务绑定

   ②kafka源算子新老API差异? 写法不同(老的创建对象,新的直接build),偏移量(老的定期自动提交offset到kafka主题,无法exactly-once,因为提交时机无法把控且无法与数据的sink端事务绑定,新版没有打开offset自动提交,但是通过把消费位移记录在状态,所以能实现精准一次,offset自动提交也可以打开用于监控系统跟踪消费偏移量)

   ③kafka源算子怎么能生成有界流? setBounded/setUnbounded控制读取的偏移量的始终,以及读完是否退出

3. source算子拓展?

   ①addsource和fromsource差异? 接收的对象不同(SourceFunction接口的实现类/Source接口的实现类),即新老api产生的不同source算子,通过不同的方法添加到env上得到流

   ②source算子作用? 从kafka某主题上一次提交的偏移量读(没有就读最新的),反序列化成String,算子添加到环境,得到起始的流

   ③source算子的并行度(是否可并行)? 对于fromElements/fromCollection这种,底层强制并行度1,就算手动设置会异常,fromParallelColllection才是多并行度source算子,还有kafkaSource以及自定义Source实现可并行SourceFunction接口,直接看Source算子类是否实现可并行接口

4. 滚动聚合算子max和maxBy区别?

   ①按道理像hive的select不可以选非聚合字段(会歧义),flink这里max可以取,取组内第一个的,同样有该现象的有mysql的非严格模式,原理:状态中保存非聚合字段为组内第一条数据的,之后更新只改聚合字段;

   ②maxby的处理? 不像max只替换聚合字段,maxby更新所有字段(整条数据),所以输出结果为最大值所在的那条数据

   ③maxby遇到和最大值相同? 不会替换原来的最大值所在数据,大于才会替换

5. 文件sink的了解?

   ①writeastext? 覆盖模式下,目标路径生成n(并行度)个文件,且无法滚动(生产环境淘汰),不覆盖模式下如果文件存在会报错

   ②writeascsv? 一样无法滚动,只能将元组流输出为csv文件,毕竟flink不知道你类型中有哪些字段,这里元组中有嵌套结构(比如Map<String,Integer>)会调用toString,因此建议转json方便下游解析

   ③流中来新数据就会往下游写出吗? 有缓存(包装流中缓冲的大小,比如4096Byte,即4k)大小限制和时间间隔

   ④数据有改动就会写出吗? 有缓存(包装流中缓冲的大小,比如4096Byte,即4k)大小限制和时间间隔
   ⑤writeas…算子底层? 调用writeUsingOutputFormat,只是输出文件格式不同,传入的OutputFormat对象不同,再底层就是IO流的输出了
   ⑥生产级文件sink? StreamFileSink,目前唯一能实现端到端精准一次
   ⑦flink结果数据输出到文件系统意义? 方便给后续离线计算系统提供数据

3 扩展sink算子

1. StreamFileSink说明?

   ①SFS的优势? 分桶(分文件夹,比如按文件大小和小时)滚动,可写出列式存储格式文件,生命周期(在写->挂起->最终完成),完成后滚动写新文件,滚动策略

   ②整体api? FileSink类调用编码方法后关联分桶策略、滚动策略、文件配置最后.build得到SFS算子,后将其作为参数传入SinkTo方法添加到流中,写出到文件

   ③按行编码/按块编码区分api? 按行编码在FileSink类调用编码方法时传入格式类和路径即可,而按块bulk编码传入的是输出器工厂,以输出列式存储文件parquet为例,这里就用到ParquetWriterFactory对象,而该对象的创建有3种模式,对应ParquetAvroWriters的三种方法,区别在于传参(schema对象/特定的javaBean/普通的javaBean)

2. StreamFileSink按bulk块编码输出列式存储文件说明?

   ①按行/按块编码有何区别? 顾名思义为编码处理的批次大小,涉及到采用压缩时的压缩率,明显后者高,同样道理,若想实现列式存储,按行编码不能实现同一列放一起,因此就得按块编码

   ②需要与checkpoint整合吗? SFS如果不开cp的话,文件一直是inprogress,无法挂起和完成,因此其作用等于事务写入,注意,一般时候cp的作用只是改变生命周期,不是滚动的时机,除非按块编码

   ③按块编码时的滚动策略? 注意,只能当cp发生时,进行文件滚动,而且无法按文件大小/时间,这是因为列式存储文件不能按大小/时间去切,整体结构嵌套复杂且浑然一体,不能切

   ④列式存储文件结构? 整体分为文件头(标记)+体+尾(元数据:索引,统计信息即字节大小和压缩前后大小等),文件体按行分为行组,一大块内按列分为小块,列存储块中分页(按量),页中分为页头等元数据+原始数据两部分,而这种结构parquet的API,flink通过ParquetWriter工厂封装API,从而SFS调用写出parquet文件,avro只是parquet选的序列化协议

   ⑤谈谈api的重点? 既然④中说到,列式存储文件如parquet,avro,orc自带完整的schema信息,不用手动赋予,因此在sink写出时,api中一定能体现结构信息schema的创建,这才有了大致步骤,FileSink算子调用编码方法传入bulkwriter块输出器工厂->应该传入parquetAvrowriter工厂,但是缺少schema->多种模式下(avro的API手动构造schema/avsc描述文件自动生成特定JavaBean/反射)定义schema,而且流的泛型需要改为特定的JavaBean

   ⑥三种模式下生成schema的区别? 注意,api不仅完成了schema的构建,而且还将流的泛型改为了特定的JavaBean,这和spark的rdd转dataframe的核心操作完全一致,即改行类型row+添加schema这两件事,但是flink这里三种模式的实现不同,模式1:Avro的Api,要schema就传入schema;模式2:avsc描述文件+插件生成特定类,调用特定类中的方法获得schema,模式3:传入普通类,通过反射获取普通javabean中字段信息翻译成schema

   ⑦一些api需注意的问题? 基本都是一些序列化的问题,因为数据写出时使用的是avro序列化器,比如流的泛型中某字段类型无法序列化,schema的某字段类型无法序列化

3. KafkaSink算子?

   连接服务+序列化(指定主题)+语义(端到端精准一次/至少一次)+事务号前缀+生产者参数

4. JdbcSink算子?

   分为精准一次和不保证精准一次两种api,区别就是有没有开启mysql的两阶段事务提交,当然前者使用幂等插入配合表的主键也能实现精准一次

4 多流操作

1. 如何实现分流?

   ①不管是过时的.split算子还是process算子侧输出流实现分流,原理都一样就是在处理数据时打上不同的标签,然后主流获取不同标签的侧输出流;

   ②一条数据能输出到多条分流吗? 可以,这是java基础的逻辑,只需将数据打上多个标签即可

   ③标签的作用? 约束侧流约束并且需要传入类型信息

2. coGroup协同分组和join关联?

   ①cogroup理解? 两条流相同key数据分到一组(协同组,一组内两个迭代器,调用方法处理一次),又因为流式处理所以需要加入窗口转为有界流才能关联上,至于关联不上的数据也是一组(输出结果与否看处理逻辑)

   ②join? 底层调用cogroup算子,算子内的join方法处理的是组内(相同key)关联上的一对数据,即inner join,DataStream API想要实现外联/左联只能通过cogroup算子

3. 广播连接流?

   ①使用场景? 可以将主流和广播流看成事实表和小维表/字典表,一个变化快比例大,一个变化慢比例小,则将广播流数据写入状态共享给主流算子所有分区,又因为广播状态按map结构存储,所以需传入映射状态描述即kv类型来构建广播流,与主流连接

   ②广播状态的相关null指针问题? 处理广播流时写入状态不会出问题,问题在于如果主流数据先到/广播流没有数据,那么在处理处理数据时读取状态就会null指针,因此需要对状态和状态中的元素做null校验
   ③为什么提供只读状态? 安全性考虑,防止主流中改变广播状态,造成子任务间状态的不统一
   ④广播连接流的理解? 应该理解为connect,而不是join,所以流式处理而非批处理,主流数据只会处理一次,状态流来数据永久存入状态

5 并行度相关概念

1. slot槽位、并行度、槽位共享的理解?

   首先若不考虑并行度,我们知道,算子将逻辑封装在task中,而task会在job执行流程的最后被JobManager部署到TaskManager中运行->开始优化:task如果是一个进程的化,为什么不开启多个线程呢?->引入并行度的概念,将task以多并行实例subtask部署到TaskManager,而TaskManager上为每个线程(subtask)提供资源的就是slot,这时一个subtask占用一个槽位->开始优化:考虑到资源浪费和不足,让多个subtask部署到同一slot,即槽位共享,但是不同算子的subtask才能共享slot->为什么又不让同一个算子的subtask分到相同的slot中呢,这是因为subtask的出发点就是并行处理,而slot的资源有限若相同算子的subtask轮流使用同一slot,那不就回到串行运行了嘛,subtask设计的意义就没有了->因此,并行度>slot数会报错->slot共享最终实现了该串行的串行(减少了网络IO和线程间切换),该并行的并行,节省了消耗并提高了处理效率

2. 算子链?

   合并算子链作为一种优化手段,原理其实就是将多个算子的逻辑封装到一个task中,这样就节省了机器资源(网络传输和线程的切换)->上下游算子满足3个条件才能合并(oneToOne+并行度相同+属于相同槽位共享组),为什么->onoToOne是为了保证合并后处理逻辑的正确性(不然不可能拿到上游所有实例计算结果),并行度相同是为了保证最终的subtask处理逻辑统一,属于相同槽位共享组subtask才能部署到同一slot

3. 分区算子?

   ①上游subtask数据会发往下游哪个task? 首先,上下游算子并行度相同则数据一对一发送forward,否则需要经过分区器,比如keyby一定走keyGroupStreamPartitioner,普通算子并行度不同也需要走分区器导致有多种数据分发策略,默认是rebalance轮询
   ②数据分发策略由谁指定? 一般都是算子自动指定,也可以手动指定(rebalance/rescale/shuffle是随机,不是spark的shuffle),注意:分发策略包含两部分(发往下游哪几个分区,分发的方式是轮询还是其他)
   ③keyby底层? KeySelector得到key,对key两次hash模以并行度,得到目标分区编号

6 时间语义、WaterMark、窗口

1. 为什么会有多种时间语义?

   数据因为网络IO以及多并行实例间的差异导致处理数据时的时间与数据产生的时间不一致,因此针对不同业务需求选取不同的时间语义

2. 为什么会有watermark?

   watetmark用于解决事件时间推进的2个问题,网络IO造成的数据乱序问题(隔200ms更新watermark广播到下游,迟到数据不会改变watermark),以及上游多并行实例造成下游时间标准不统一的问题(谁慢以谁为准避免产生大量迟到数据)

3. 算子的watermark用什么更新,有什么难处?

   ①看你定义watermark产生的源头位置,一般在source算子处定义产生,source的任务实例用已到数据中最大的事件时间更新,其下游所有算子用上游发送的watermark推进时间

   ②水印从上游广播到下游,那么问题来了,下游一个subtask收到上游多个subtask的水印,以谁为准? 为了避免产生大量迟到数据,选择以上游水印中最小的来更新

   ③既然以上游最慢的水印为准,那么问题来了,如果上游某一subtask的没有新数据来即水印无法更新,下游窗口算子岂不是永远无法关闭并触发计算了,还会有数据积压等问题? 设置超时参数,指定watermark推进检测间隔,若上游某分区迟迟没有推进则主动推进下游的整体watermark以关闭后续窗口

4. 单并行度下观察watermark传递及问题分析?

   ①首先,precess处编写watermark打印代码,当没有数据的输入,watermark在算子间以极小值传递却无法被观察到,第一条数据2000时间输入,process打印的是极小值即上一次更新的,上游的1999watermark紧随2000数据之后,处理完2000数据后更新process的watermark为1999

   ②在方法中,打印watermark前让线程阻塞1s,新数据到来,为什么结果仍不是最新数据的watermark? 按道理水位线200ms间隔更新,因此有2种可能:watermark已异步更新但是上下文是1s前的上下文/200ms异步更新无效,即200ms更新为串行而非异步(往下看,证实2种猜想都错了)

   ③查看水位线的更新代码? 先更新后发射,断点+周期,断点方法OnEvent为立即使用新输入数据更新最大时间戳,周期方法(异步)为:TimeStamp&Watermarks算子的定时器200ms间隔调用onperiod方法发射内容为最大时间戳的水印(没有数据也会发射,比如一开始的极小值)

   ④ 在③中可以看到,新数据到来,水位线虽算是更新了但是并没有立即发射,而是等待发射间隔后发射,那么新数据的发送和水位线的更新谁先谁后? TimestampsAndWatermarksOperator算子代码说明,先收集数据往下游输出,再调用onEvent方法更新本地水印

   ⑤ 通过③和④总结时间戳水位算子的执行内容? 首先,定时器周期性异步触发OnPeriodEric方法发射水印(最大时间戳-1),水印算子收到数据,先往下游发送,再更新maxtimestamp,再周期性发射新水印

   ⑥回到②的问题,process算子先收到新数据,再在200ms内收到了新数据的水印,那为什么打印的还是老水印呢? 下游process算子是串行方法判断数据类型(数据/水印),若是数据则调用process算子逻辑处理,若是水印则更新channelstatus(多分区则有多个,则用最小的更新,即多分区水位线更新原则)中watermark, 回到②中问题,新数据先到新水印后到,因此处理数据时水印还未更新,就算sleep一天新水印也未被接收并更新,所以数据处理方法中打印的是旧水印,之后才会处理后接受到的新水印

5. 窗口相关概念?

   ①窗口分类? 按时间/条数,按滑动/滚动/会话,按是否KeyedStream
   ②KeyWindow和NonKeyedWindow区别? 前者有并行度,后者并行度为1
   ③窗口触发器,擦除器,allowlateness? 窗口触发器为窗口关闭执行触发计算,擦除器为窗口关闭擦除桶内部分数据(滚动窗口则擦除桶内所有数据),allowlateness为窗口关闭在允许范围内只触发,不擦除,迟到数据进入桶并触发,允许范围过后才擦除(如果是滑动窗口则会影响后续窗口结果正确性)
   ④窗口处理乱序数据总结? 小乱序:利用水印容错时间,防止产生大量迟到数据;中乱序:利用窗口允许迟到机制,延迟窗口关闭;大乱序:利用迟到数据侧流输出机制(针对窗口擦除器执行之后的数据)

   ⑤窗口函数api? reduce和aggregate等聚合算子传入的滚动聚合函数,apply和process传入的全窗口函数

   ⑥选哪种窗口函数好? 1看需求,如果需要拿到所有数据才能得到结果选全窗口函数比如排序,否则用滚动聚合函数,滚动的是当前结果来一条数据滚动更新一次,2看优势,滚动的效率高,全量的逻辑可以很复杂

   ⑦key怎么和keyedwindow绑定的? keyedwindow算子有并行度,虽然窗口算子拿到一个分区数据但是可能包含多种key,但是一个窗口只会统计自己key的数据,这是因为key的hash实现key和窗口绑定

   ⑧滑动窗口和滚动窗口的关系? 滚动是步长为长度的滑动,滚动中一条数据只属于一个窗口,滑动一条属于多个窗口

   ⑨计数窗口算子vs时间窗口? 底层调用的还是window算子加上了触发器和移除器,相当于做了特殊的封装,因此底层调用的算子和时间窗口底层相同只是触发器和移除器不同

6. 触发器、移除器?

   ①事件时间窗口触发器? onElement:事件数据到达判断水印超过窗口的maxTimeStamp(end-1)则开火,否则定时器注册maxTimeStamp继续攒数据,onEventTime则处理到达的水印与定时器的时间比较判断是否触发,而OnProcessingTime方法则没有写处理逻辑
   ②计数窗口触发器? 类似的,主逻辑在onElement方法,计数值超过窗口长度则触发,同时计数归零,否则继续攒
   ③移除器? 窗口触发前后,对桶中的数据移除(移除上一个窗口的数据),即触发计算的铺垫操作
   ④事件时间窗口移除器TimeEvictor? 默认在触发前移除,对pane中数据,即迭代器,取最大时间戳,减去窗口长度得到移除边界(滑动窗口则是减步长),遍历迭代器中数据同时移除边界前的数据,即上一个窗口的数据(当然也可以自定义移除本窗口的部分数据)
   ⑤自定义触发器? 借鉴EventTimeTrigger代码,自定义类继承其父类,onEventTime方法中编写处理事件逻辑
   ⑥自定义移除器? 若需要移除触发标记数据a,则遍历迭代器时判断为a即可,至于a可通过构造器传入更灵活,api需继承TimeEvictor的父类,否则直接继承TimeEvictor会导致size变量歧义
   ⑦这里移除器移除a那么触发器则无法遇到a又如何触发呢? 其实,a数据已经触动触发器只是移除数据逻辑可以先于计算处理桶数据的逻辑,即可以理解为触发器包含移除和窗口计算2部分逻辑

7 Flink状态

1. 状态和普通变量有什么区别?

   flink state(托管状态)/raw state(自管理状态),其实所有状态都可以等价在算子传入类中自定义变量实现,比如值状态的功能就等价于通过定义普通变量保存临时计算结果,但是无法实现自动管理即容错,而手动定期持久化和恢复的代码逻辑又过于复杂,不如flink的状态管理器直接封装成api调用简单

2. 算子状态与键控状态区别?

   前者的状态与算子绑定,虽然不同subtask的状态不同,但是这里不能说算子的状态与subtask绑定,因为这是数据的分配策略造成的(默认轮询),同时算子状态恢复时,状态并不能回到原来的subtask而是均匀分配,包括恢复时并行度变化也是状态在subtask间均匀分配,后者状态与key绑定,这种绑定原理与窗口和key的绑定类似,一个subtask接收若干个key的数据并维护其对应的状态,因此多并行度也不会导致subtask间键控状态混淆紊乱

3. 初步认识状态与checkpoint?

   ①绑定的,只有状态没有cp,无法持久化和恢复状态,使得状态初始化时无法获得之前的状态

   ②算子状态api? 状态相关:自定义算子类继承算子类并实现CheckpointedFunction接口,处理数据方法中更新状态(拿到状态为复杂数据结构,而不是简单的List对象,所以先迭代器获取状态内容),初始化方法中初始化状态数据(存储器+描述器),checkpoint相关:快照周期+模式+路径,默认只保留最近一次的快照(即覆盖)

   ③这里程序重启为什么没有初始化得到之前的状态呢? 首先得区分job级别失败和task级别失败,task失败则failover机制(需代码指定开启自动重启策略,否则task失败也会导致job失败)自动让其重启,算子类中状态初始化方法自动加载快照数据,而程序重启为job失败不会加载快照,需要加载的话必须人工指定,即状态恢复只支持task级别的,job级别容错需手动处理

   ④ 在③中task重启期间处理数据是否会丢失? 会,因此需要数据重放处理

   ⑤键控状态api? 算子状态使用必须实现CP接口,在初始化方法中拿到状态管理器以初始化,而keyed状态直接可以在rich的算子类中拿到和初始化

4. 对比键控状态和算子状态,思考:键控状态和算子状态到底谁能实现字符串拼接?

   键控状态:数据按键分区,状态按key在分区内隔离,就算并行度为1也无法合并所有key的状态,因此不行(除非将所有数据定死相同key);算子状态:只有并行度为1,才行,否则无法合并多个subtask的状态

5. 算子状态恢复的重分配问题?

   如果任务失败重启时,算子并行度改变,则原来的状态数据会汇总后重新均匀分配(轮询)给subtask,一般用于source算子(其他算子使用会因为计算逻辑而影响结果正确性),比如kafkaSource,并行度由原来的和kafka分区数对应在task重启时改为降低,则某一subtask读取原来2个subtask状态(偏移量)同时改为消费2个kafka分区,键控状态由于key和状态绑定所以不存在重分配问题;list和unionList状态区别就在于任务重启后,加载状态数据时,重分配策略不同,前者均匀分配(轮询),后者广播

6. 状态TTL管理?

   ①ttl配置中状态模式理解? 场景:hbase中画像的数据高频地读,读hbase操作重因此需要缓存,即存入flink状态,查询越多状态存储空间越大,占用内存且影响查询效率,因此考虑删除状态中冷数据,因此将默认的不更新改为onCreateAndWrite(写刷新存活间)/OnReadAndWrite(读写刷新),实现冷数据的ttl
   ②ttl对过期数据的处理? 过期状态数据默认虽不可见(类似移除器,查询前先过滤过期数据),但会异步清理(可以参数设置多种清理策略),因此可以手动配置过期但未清理的状态数据为可见
   ③ttl的时间语义? 默认是处理时间语义,不支持别的语义
   ④ttl配置的理解? 关于ttl的所有配置都是对状态中的每条数据,比如ListState的每条数据有自己的存活时间和刷新,而非整个状态内容

7. 状态后端?

   ①功能:状态快照,快照保存,保存格式,快照恢复,状态过期清理等
   ②前置知识:rocksdb类似derby,是一种嵌入式数据库,不像derby是支持sql的关系型数据库,它是kv数据库
   ③HashMapStateBackend(默认):运行时状态保存在堆内存中,数据以对象形式存在,比如List对象/Map对象,多则溢出到本地磁盘
   ④RocksDbStateBackend:数据库内嵌在taskManager中,状态数据序列化成字节后写入数据库,同样先放内存多则溢出到本地磁盘,
   ⑤对比:主要就是对比读写效率,前者内存中读写快而一旦溢出磁盘则读取时无法使用数据结构和算法效率自然下降,而后者虽需序列化和反序列化导致效率不如前,但是磁盘数据文件的检索在数据库中利用特有的数据结构设计会得到提升,因此超大规模状态需用后者
   ⑥注意:两种状态后端虽运行时状态(本地状态)保存格式差异大,但是状态快照到hdfs上的文件格式等却相同,这也是为了敏捷开发方便切换状态后端
   ⑦总结:如何选择状态后端,其实是就是对比内存资源和状态的大小

8. 状态TTL清理策略?

   ①增量清理(默认):仅对HashMap状态后端生效,手动遍历整个状态内容(包含过期状态数据a),a才会被清理,即获取a并发现其已过期才会清理a,由于遍历和获取的资源开销大因此不可行
   ②全量快照清理:状态做快照时,需要清理过期数据,cp完成后快照中保留的就是清理后的数据,但是本地的状态仍未清理过期数据
   ③rocksdb压紧:仅对RocksDb状态后端生效,利用RocksDb的compact机制(类似hbase)清理过期状态数据
   ④问题:虽然显式设置清理策略能实现,比如增量清理,但是不设置时清理结果却对不上默认的清理策略? 这是因为,默认是传参为(5,false)的增量清理策略,首先key绑定的状态存于subtask的copyOnWriteStateMap数据结构中,而开辟的空间大小类似map先128再扩容,默认的增量清理策略的传参中传入size和boolean,size是访问状态时迭代器(包含所有key)检查多少个key的过期数据,因此只会清理检查到的5个key绑定的状态中过期数据而后推进,所以就算某个key的状态访问过但因为不在5个中所以不会清理,比如key为a在map中的第22条,则第5次清理才轮到a,而传入true则每条都检查造成延迟

8 CheckPoint机制

1. checkpoint的难题和核心机制?

   分布式全量快照核心就是保证所有算子处理完同一条数据才快照(插入barrier,防止前后算子处于不同轮快照,恢复时影响逻辑),再配合source偏移量重放数据(要么全不要用上一次的快照,要么用最近的快照),才能实现端到端至少一次

2. checkpoint流程?

   也是2PC(两阶段协议),JobManager定期插入barrier到source,各个算子完成本地快照(cp-n)会返回ack,JM收到所有算子应答才认为本次cp全局快照完成,随后JM向各个算子通报(notify/callback)cp-n完成

   思考checkpoint过程中前后算子的本地快照并非同一轮怎么办? 下游算子收到第一轮barrier同时上游已经收到第二轮barrier,上游算子有2个版本快照不会混乱吗?回答是下面流程中算子的应答内容中都带有cp编号的,不会混淆

3. barrier对齐和非对齐的checkpoint,及各自的危害?

   ①barrier对齐? 做本地快照时,上游算子并行度高,barrier因为时延不同导致到达下游时间不同,只能等最慢的barrier到来,因此快的流会有下一次的cp数据到来,只能将这些数据搁置缓存中而不影响状态,直到barrier对齐开始本地快照(异步),同时处理积压数据
   ②对齐的危害? 影响延时和效率,数据积压,背压往上游传递直到数据源头
   ③非对齐的checkpoint? 保存本地状态的同时保存上下文环境,虽不会产生背压但只能实现至少一次语义

4. checkpoint的api?

   cp的eos/atleast once语义其实就是对齐/非对齐,可以不默认清除最后一次cp的数据(可以人工实现job无缝连接,即job重启读取之前cp状态数据),快照其实是周期且异步的,所以可以设置最大并行cp数防止背压等问题,可以设置cp最小间隔防止cp占用subtask处理数据时间

9 容错机制

1. 简述flink各环节状态一致性保证?

   ①source保证? 可记录offset+可重放+offset可保存在状态中同下游算子状态一起快照
   ②flink内部eos语义保证? 分布式快照算法,插入barrier,将流逻辑上分为2部分,为保证cp后所有算子都是只经过了相同数据的一次影响,确保要么保留正确处理一批数据后的状态,要么中途失败回到未处理这批数据的状态
   ③sink保证? 幂等写入/两阶段写入(预写日志方式在将状态中的数据写入时崩溃,由于offset和事务已提交,则数据丢失)

   ④如何理解flink的精准一次? 并不是一条数据只会处理一次,而是状态数据最终效果就只受数据的一次影响

2. sink端容错?

   ①幂等写入? 幂等写入只能实现最终一致性(存在跳回重放现象,即过程中不一致),kafka支持的是producer和brokers间的幂等写入(消息的序列号),而flink写入kafka不支持幂等写入,因为这属于生产者发送重复消息而非发送消息到brokers时失败重试,因此flink中产生重复的数据(比如leftjoin)写入kafka存在去重问题

   ②两阶段事务写入? sink两阶段事务写入利用的就是checkpoint的两阶段提交协议(算子收到barrier+收到notify),对应的就是sink算子预提交阶段(开启事务,处理数据,输出数据-这里输出的是barrier后面的数据,本地状态快照)+提交阶段(提交事务-这里提交的是barrier前面的数据)

   ③两阶段预写日志提交? 当外部系统不支持事务/幂等性(flink在内部模拟事务),写出数据存入sink算子的状态直接当作预写日志,等收到checkpoint完成的notify,再写出,对比事务的优点:数据不会残留在外部系统且不要求目标系统支持事务

   ④幂等写入对比事务写入? 及时性对比,一个实时写入一个攒批写入

   ⑤kafka事务写入? 伪事务,开启事务后写入的数据做标记,无法被消费,提交后才能正常消费;

   ⑥sink中也会有状态吗,也需要快照吗? sink的状态并非等于写出的数据,而是与算子逻辑有关,也和写出方式有关(③中待写出数据就存入状态),任何算子都可以有状态,而且需要在checkpoint时快照
   ⑦两阶段事务提交数据会丢重吗? 正式提交事务时挂掉导致数据丢失(因为cp完成flink认为事务已经提交不会数据重放),解决办法:下游通过事务id处理未提交的事务,或者task重启时读取sink中预提交阶段存入状态的事务信息,发送给外部系统将pending的事务完成finished(因此对系统有要求)并修改事务信息为finished(如果提交完但信息未修改就挂了会导致数据重复)

   ⑧预写日志提交存在的问题? 数据会重复写出

   ⑨关于框架的数据丢重? 个人理解,一旦某个数据io环节之后没有ack返回操作,即该io环节没有与前面的操作进行原子绑定,那么协调者就无法判断该io是否成功也就无法补救,最终造成数据丢失,反过来先io再返回ack则会导致数据重复写出,因此丢重客观上一定存在
   ⑩sink两阶段事务提交勘误? 预提交阶段:开始事务+输出数据+barrier到达+预提交事务(存储本次对外事务号,及事务状态:pending)+做本地状态快照+上报+等notify;提交阶段:收到notify+提交事务(对外commit,成功则修改事务状态:finished)

3. task级失败重启恢复状态?

   ①Task级别的故障重启,是系统自动进行的,但可以手动配置策略

   ②自动重启策略? 1.固定上限延时重启,超过上限job失败;2.默认的不重启;3.奖惩间隔重启;4.失败比例重启;5.重启策略在配置文件flink-conf.yaml文件中

   ③重启是自动重启所有task还是部分? 看failover策略.默认pipelined region:重启故障task关联的region最小集即pipeline(举例:整条流都是oneToOne关系算子,并行度相同的就可以不重启全部);all:重启所有task

4. cluster级失败重启恢复状态?

   ①不同于task级别失败自动重启会自动读取cp数据恢复状态,job失败不会自动重启,且默认删除cp数据且不恢复状态,因此作业恢复需用到flink的另一快照备份机制savepoint

   ②savepoints对比checkpoint? cp自动周期性触发,savepoints其实是人工的cp而后手动命令重启job通过cp/savepoints数据恢复cluster状态

   ③恢复状态? 方式一:手动执行命令生成savepoint文件,job重启时命令行参数指定从savepoint文件恢复之前的cluster状态;方式二:环境中设置job结束时不删除checkpoint文件,则job重启时指定从checkpoint文件恢复cluster状态

10 Flink分布式部署

1. 对比spark的分布式?

   ①spark的rdd工作流程? 子rdd的compute方法,调rdd的迭代器,形成链,数据从前往后应用函数,阶段划分,数据往文件(shufflestage)/输出器(resultstage)写
   ②sparksql执行计划? sql->语法树->解析树(逻辑执行计划)->逻辑优化树->物理执行计划(树)->RDD层级代码(代码生成)
   ③spark的driver理解? 一个线程,调用的是用户类的main方法
   ④spark on yarn的client/cluster模式区别? 就是用户类的main方法在哪里运行的区别,前者在客户端job提交进程里调用,后者在AppMaster进程中调用,即,该main方法在哪里调用,对应的dag生成stage划分task生成和调度就在哪里完成,这就说明客户端在前者模式下不能退以保证持续发送task同时造成客户机压力大

2. standalone集群?

   ①job提交流程? env.execute->客户端生成流图(点为算子,边为流向),加工(合并算子链等)成job图,发给集群的jobmanager->jobmanager对job图加工(结合算子并行度,给图加点和边)成(逻辑)执行图->执行图中各任务在slot上执行,得到物理执行图

   ②web界面? 可以看到节点的线程繁忙/背压程度,有些分区虽然没收到数据但是会有元数据(barrier,水印等)因此也有收到字节

   ③提交模式? -d为分离模式即提交完job退出client,否则为session模式client不退出

3. Flink on Yarn?

   ①三种模式? session(job共享资源,main在client执行)/per-job(资源独享,client)/application(独享,cluster)

   ②上述3种模式的区别? 集群的生命周期和资源的隔离保证;用户类的main方法是运行在client端,还是在集群端

   ③session模式命令行操作? 先session命令起集群(注意这里仅向yarn申请容器启动jobmanager,却没有指定taskmanager个数因此比独立模式有弹性),然后run命令提交job

   ④application模式的优势? job提交后,main需要执行,因此需要生成图,而per-job在客户端解析消耗该服务器算力,application在集群解析因此更适合企业多人开发场景

4. standalone集群对比Flink on Yarn?

   独立集群缺点? session模式,资源层面(利用弹性不够+隔离度不够即共享严重负载大)

   如何理解独立/on yarn? jobmanager和taskmanager进程运行在哪,在服务器操作系统上就是独立,在yarn提供的容器就是onyarn

   注意区别standalone集群和Flink on Yarn下session模式的不同? 不同就在于taskmanager的弹性,flink on yarn下先启动session服务就等于先启动jobmanager,而taskmanager是在job提交时按需动态启动的,而standalone集群的taskmanager是提前按slave数启动好的