企业qq免费版,安徽建站优化哪里有,wordpress网站描述,沈阳做企业网站Flink面试题目合集
从牛客网上找到的一些面试题#xff0c;如果还有其他的#xff0c;欢迎大家补充。 1、能否详细描述下Apache Flink的架构组件和其工作原理#xff1f;请介绍一下Flink on YARN部署模式的工作原理。
官网图#xff1a; 由两个部分组成#xff0c;JM如果还有其他的欢迎大家补充。 1、能否详细描述下Apache Flink的架构组件和其工作原理请介绍一下Flink on YARN部署模式的工作原理。
官网图 由两个部分组成JMTM。 JM中包含三个组件dispatch、jobmaster、resource manager。 dispatch主要是负责提供了rest接口接受客户端提供的jar包、dataflow、streamgraph等信息并且运行Flink UI也是该组件运行。 jobmaster主要是负责将dataflow等数据流图转换成真正的物理执行图如果资源足够启动任务那么就会将图分发给对应的TM并且负责任务启动后的协调运行管理比如checkpoint的协调。 resource manager主要负责资源的申请、释放、分配。管理着Flink的slot。
TM主要是任务执行的地方根据slot的个数决定启动的task线程个数。其中每一个TM共享着NetWorkBufferPool、NetWorkEnvironment。NetWorkBufferPool管理着TM的可用的内存MemorySegment默认是32k。每一个slot线程任务会存在输入区域inputgate和输出区域result partition对应两个local buffer pool这个是根据NetWorkBufferPool进行动态平均分配的。详细了解Flink 解析二反压机制解析_flink的taskmanager内反压过程-CSDN博客 2、Flink的窗口操作有哪些类型它们之间有什么不同请举例说明如何定义不同类型的窗口。
直接盗用官网例子。。。
Keyed Windows
stream.keyBy(...) - 仅 keyed 窗口需要.window(...) - 必填项assigner[.trigger(...)] - 可选项trigger (省略则使用默认 trigger)[.evictor(...)] - 可选项evictor (省略则不使用 evictor)[.allowedLateness(...)] - 可选项lateness (省略则为 0)[.sideOutputLateData(...)] - 可选项output tag (省略则不对迟到数据使用 side output).reduce/aggregate/apply()[apply方法已过时一般使用process下同] - 必填项function[.getSideOutput(...)] - 可选项output tag
Non-Keyed Windows
stream.windowAll(...) - 必填项assigner[.trigger(...)] - 可选项trigger (else default trigger)[.evictor(...)] - 可选项evictor (else no evictor)[.allowedLateness(...)] - 可选项lateness (else zero)[.sideOutputLateData(...)] - 可选项output tag (else no side output for late data).reduce/aggregate/apply() - 必填项function[.getSideOutput(...)] - 可选项output tag滚动窗口
基于时间
// 滚动 event-time 窗口
input.keyBy(key selector).window(TumblingEventTimeWindows.of(Time.seconds(5))).windowed transformation(window function);// 滚动 processing-time 窗口
input.keyBy(key selector).window(TumblingProcessingTimeWindows.of(Time.seconds(5))).windowed transformation(window function);// 长度为一天的滚动 event-time 窗口 偏移量为 -8 小时。
input.keyBy(key selector).window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(-8))).windowed transformation(window function);滑动窗口
// 滑动 event-time 窗口
input.keyBy(key selector).window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5))).windowed transformation(window function);// 滑动 processing-time 窗口
input.keyBy(key selector).window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5))).windowed transformation(window function);// 滑动 processing-time 窗口偏移量为 -8 小时
input.keyBy(key selector).window(SlidingProcessingTimeWindows.of(Time.hours(12), Time.hours(1), Time.hours(-8))).windowed transformation(window function);
如上例子所示窗口的 assigners 也可以传入可选的 offset 参数。这个参数可以用来对齐窗口。 比如说不设置 offset 时长度为一小时的滚动窗口会与 linux 的 epoch 对齐。 你会得到如 1:00:00.000 - 1:59:59.999、2:00:00.000 - 2:59:59.999 等。 如果你想改变对齐方式你可以设置一个 offset。如果设置了 15 分钟的 offset 你会得到 1:15:00.000 - 2:14:59.999、2:15:00.000 - 3:14:59.999 等。 一个重要的 offset 用例是根据 UTC-0 调整窗口的时差。比如说在中国你可能会设置 offset 为 Time.hours(-8)。基于计数也分滚动和滑动
stream.keyBy(可选).countWindow(size)会话窗口
// 设置了固定间隔的 event-time 会话窗口
input.keyBy(key selector).window(EventTimeSessionWindows.withGap(Time.minutes(10))).windowed transformation(window function);// 设置了动态间隔的 event-time 会话窗口
input.keyBy(key selector).window(EventTimeSessionWindows.withDynamicGap((element) - {// 决定并返回会话间隔})).windowed transformation(window function);// 设置了固定间隔的 processing-time session 窗口
input.keyBy(key selector).window(ProcessingTimeSessionWindows.withGap(Time.minutes(10))).windowed transformation(window function);// 设置了动态间隔的 processing-time 会话窗口
input.keyBy(key selector).window(ProcessingTimeSessionWindows.withDynamicGap((element) - {// 决定并返回会话间隔})).windowed transformation(window function);
全局窗口
DataStreamT input ...;input.keyBy(key selector).window(GlobalWindows.create()).windowed transformation(window function);
3、请详细介绍Apache Flink中的Watermark水位线机制。实现Watermark需要哪个接口应在哪里定义其主要作用是什么
watermark是继承了StreamElement专门触发EventTime窗口计算其本质其实就是一个时间戳。Watermark 是和事件一个级别的抽象其内部包含一个成员变量时间戳timestamp标识当前数据的时间进度。Watermark实际上作为数据流的一部分随数据流流动。如果存在多个数据源时Flink内部为了保证watermark保持单调递增Flink会选择所有流入的EventTime中最小的一个向下游流出。从而保证watermark的单调递增和保证数据的完整性。
目前Flink有两种生成watermark的方式
Punctuated通过数据流中某些特殊标记事件来触发新水位线的生成。这种方式下窗口的触发与时间无关而是决定于何时收到标记事件即数据流中每一个递增的eventTime都会产生一个Watermark。在实际的生产中Punctuated方式在TPS很高的场景下会产生大量的Watermark在一定程度上对下游算子造成压力所以只有在实时性要求很高的场景才会选择Punctuated的方式生成watermark。 Periodic周期性的(如一定时间间隔或者达到一定的记录条数)产生的一个Watermark。在实际的生产中Periodic的方式必须结合时间和积累条数两个维度继续周期性产生Watermark否则在极端情况下会有很大的延迟。
如果需要自定义watermark生成核心是需要实现WatermarkGenerator接口需要new WatermarkStrategy,然后重写里面的createWatermarkGenerator方法return 返回我们自己实现的方法。
/*** {code WatermarkGenerator} 可以基于事件或者周期性的生成 watermark。** pb注意/b WatermarkGenerator 将以前互相独立的 {code AssignerWithPunctuatedWatermarks} * 和 {code AssignerWithPeriodicWatermarks} 一同包含了进来。*/
Public
public interface WatermarkGeneratorT {/*** 每来一条事件数据调用一次可以检查或者记录事件的时间戳或者也可以基于事件数据本身去生成 watermark。*/void onEvent(T event, long eventTimestamp, WatermarkOutput output);/*** 周期性的调用也许会生成新的 watermark也许不会。** p调用此方法生成 watermark 的间隔时间由 {link ExecutionConfig#getAutoWatermarkInterval()} 决定。*/void onPeriodicEmit(WatermarkOutput output);
}
4、能否描述Flink的窗口实现机制是如何工作的
答案同上。
5、请解释Flink的Checkpoint机制它是如何工作的Flink进行Checkpoint时整个流程是怎样的Flink的Checkpoint机制主要用来实现什么功能请描述Flink的checkpoint barrier机制。请描述Flink的分布式快照算法及其工作原理。Flink是如何保证数据在处理过程中不丢失的
checkpoint主要是全局性的轻量级快照保存的是所有算子的状态存储于状态后端主要用于故障恢复中。主要流程是同步快照、异步上传。通过JM中的checkpoint协调器从source端将barrier插入到数据流中前提是需要开启checkpoint默认不开启当barrier流经算子时就会触发该算子的checkpoint暂停计算然后保存状态然后异步上传到状态后端中算子完成checkpoint后就会将barrier广播到下游从source端到sink端都完成checkpoint才算是完成一次完整的checkpoint通知给JM的checkpoint协调器更新状态。如果是多数据源的情况下在配置为barrier对齐时算子会等待上游的所有的barrier都到齐了才会开始进行checkpoint否则就会将先到的数据流中的数据进行保存不会进行计算极端情况容易导致OOM以致于反压若配置为非对齐时当上游第一个barrier到达算子时就会触发checkpoint并且将barrier提前向下游发送同时除了保存当前状态还需要保存其余数据源的迟到的barrier之前还未处理的数据以及暂未向下游发送的数据相当于用空间换取时间。下图是非对齐barrier。 6、Flink的Checkpoint机制是如何在底层实现的Savepoint和Checkpoint有什么区别你了解Flink的Savepoint机制吗它与Checkpoint有何不同
具体checkpoint实现的源码Flink-checkPoint机制 | 智能后端和架构
实在是看不过来。。。
从概念上讲Flink 的 savepoints 与 checkpoints 的不同之处类似于传统数据库系统中的备份与恢复日志之间的差异。
Checkpoints 的主要目的是为意外失败的作业提供恢复机制。 Checkpoint 的生命周期 由 Flink 管理 即 Flink 创建管理和删除 checkpoint - 无需用户交互。 由于 checkpoint 被经常触发且被用于作业恢复所以 Checkpoint 的实现有两个设计目标i轻量级创建和 ii尽可能快地恢复。 可能会利用某些特定的属性来达到这个目标例如 作业的代码在执行尝试时不会改变。 在用户终止作业后会自动删除 Checkpoint除非明确配置为保留的 Checkpoint。Checkpoint 以状态后端特定的原生的数据格式存储有些状态后端可能是增量的。 尽管 savepoints 在内部使用与 checkpoints 相同的机制创建但它们在概念上有所不同并且生成和恢复的成本可能会更高一些。Savepoints的设计更侧重于可移植性和操作灵活性尤其是在 job 变更方面。Savepoint 的用例是针对计划中的、手动的运维。例如可能是更新你的 Flink 版本更改你的作业图等等。 Savepoint 仅由用户创建、拥有和删除。这意味着 Flink 在作业终止后和恢复后都不会删除 savepoint。Savepoint 以状态后端独立的标准的数据格式存储注意从 Flink 1.15 开始savepoint 也可以以后端特定的原生格式存储这种格式创建和恢复速度更快但有一些限制。 7、在Flink中Checkpoint超时可能是由哪些原因造成的
1、网络原因抖动导致偶发性的失败一般都会进行失败重启需要代码里面配置
2、checkpoint timeout时间不合理比如timeout的时间给太短了或者是timeout给太长导致sink的时候数据一直堆着没有提交。
3、数据倾斜或者是反压。如果是数据倾斜需要判断监控当中的task的任务状态。如果是反压那就是有可能计算资源给的不够或者看看代码中是否存在优化点。
优化方案Flink优化——数据倾斜二-CSDN博客
如果还有其他原因欢迎大家补充。。。
8、Flink如何保证Exactly-Once处理语义请解释Flink端到端的Exactly-Once处理语义并描述如何实现。Flink是如何保证数据处理的一致性的Flink任务如何实现端到端的数据一致性
通过checkpoint机制和两阶段提交保证的精准一次性。 其中flink内部是通过checkpoint将算子计算的状态定时保存到持久化存储中如hdfs或者rockdb在故障恢复时恢复到算子上一次成功保存的状态重新开始计算避免数据重复计算以及数据丢失。
两阶段提交流程当数据来了之后会开启事务正常写入但是标记为未提交预提交当barrier到达了sink算子并且完成checkpoint后JM收到了所有任务的通知开启下一阶段的事务并且发出确认信息表示当前阶段的checkpoint已经完成。sink端收到后开始正式提交预提交的数据。 如果是端到端保持一致性外部应用数据输入到source要保证Exactly-Once语义。比如kafka可以重置offsetsink到外部应用时要么保证事务要么保证幂等性。
9、Flink中的水印Watermark有哪几种类型它们有什么区别
周期性水印Periodic WaterMark如一定时间间隔或者达到一定的记录条数间歇性水印Punctuated Watermark通过数据流中某些特殊标记事件来触发新水位线的生成
真正在底层生成的水印的方法
/*** {code WatermarkGenerator} 可以基于事件或者周期性的生成 watermark。** pb注意/b WatermarkGenerator 将以前互相独立的 {code AssignerWithPunctuatedWatermarks} * 和 {code AssignerWithPeriodicWatermarks} 一同包含了进来。*/
Public
public interface WatermarkGeneratorT {/*** 每来一条事件数据调用一次可以检查或者记录事件的时间戳或者也可以基于事件数据本身去生成 watermark。*/void onEvent(T event, long eventTimestamp, WatermarkOutput output);/*** 周期性的调用也许会生成新的 watermark也许不会。** p调用此方法生成 watermark 的间隔时间由 {link ExecutionConfig#getAutoWatermarkInterval()} 决定。*/void onPeriodicEmit(WatermarkOutput output);
}
10、请解释Flink中的时间语义并讨论其在事件时间处理中的重要性。
摄入时间数据刚到达机器的时间
事件进入Flink的时间即在source里获取的当前系统的时间后续操作统一使用该时间不需要指定watermarks的生成方式(自动生成)弱点不能处理无序时间和延迟数据
处理时间数据进行计算的机器时间
执行操作的机器的当前系统时间(每个算子都不一样)不需要流和机器之间的协调优势最佳的性能和最低的延迟弱点不确定性容易受到各种因素影响(例如event产生的速度、到达flink的速度、在算子之间传输速度等)压根就不管顺序和延迟
事件时间数据本身的业务时间
事件生成的时间在进入Flink之间就已经存在可以从event的字段中抽取必须指定watermarks的生成方式优势确定性在乱序、延时或者数据重复等情况下都能给出正确的结果弱点处理无序事件时性能和延迟受到影响
综上所述
性能ProcessingTimeIngestTimeEventTime延迟ProcessingTimeIngestTimeEventTime确定性ProcessingTimeIngestTimeEventTime
11、请描述Flink中的背压机制并说明如何通过监控来识别和解决背压问题。Flink是如何处理反压Backpressure问题的请解释背压Backpressure是什么以及如何在Flink中处理背压问题。
背压机制主要是因为消费速度远远小于生产速度导致数据积压OOM最终导致任务失败。主要有两个部分跨TaskManager的反压过程和TaskManager内的反压过程。在跨TM的反压中在1.5版本之前如果存在背压情况采用的是TCP多路复用所以可能会导致TCP的通道被占用从而影响其他正常运行的任务。在1.5版本之后则是基于credit在向下游传递数据之前会先发送backlog告知下游准备发送多少数据下游则会计算剩余的buffer空间如果内存不足则会告知上游最多接受多少数据防止TCP通道被占用。好处基于credit的反压过程效率比之前要高因为只要下游InputChannel空间耗尽就能通过credit让上游ResultSubPartition感知到不需要在通过netty和socket层来一层一层的传递。另外它还解决了由于一个Task反压导致 TaskManager和TaskManager之间的Socket阻塞的问题。 TM内的反压每一个TM中都会共享一个network buffer poolTM中的task的输入区域和输出区域也会分别对应一个local buffer pool会被分配内存块进行数据传输如果存在反压那就是从输出区域开始输出区域可使用的内存块全部被用掉而输入区域的数据还在源源不断的写入积压最终导致输入区域的内存块也被用掉了最终形成反压。处理反压
首先根据Flink UI监控指标判断是什么阶段形成的反压。有可能以下情况造成系统资源 首先需要检查机器的资源使用情况像CPU、网络、磁盘I/O等。如果一些资源负载过高就可以进行下面的处理 1、尝试优化代码 2、针对特定资源对Flink进行调优 3、增加并发或者增加机器垃圾回收 性能问题常常源自过长的GC时长。这种情况下可以通过打印GC日志或者使用一些内存/GC分析工具来定位问题。CPU/线程瓶颈 有时候如果一个或者一些线程造成CPU瓶颈而此时整个机器的CPU使用率还相对较低这种CPU瓶颈不容易发现。比如如果一个48核的CPU有一个线程成为瓶颈这时CPU的使用率只有2%。这种情况下可以考虑使用代码分析工具来定位热点线程。线程争用 跟上面CPU/线程瓶颈问题类似一个子任务可能由于对共享资源的高线程争用成为瓶颈。同样的CPU分析工具对于探查这类问题也很有用。负载不均 如果瓶颈是数据倾斜造成的可以尝试删除倾斜数据或者通过改变数据分区策略将造成数据的key值拆分或者也可以进行本地聚合/预聚合。 上面几项并不是全部场景。通常解决数据处理过程中的瓶颈问题进而消除反压首先需要定位问题节点瓶颈所在然后找到原因寻找原因一般从检查资源过载开始。
12、Flink如何解决数据处理中的延迟问题
1、事件时间产生的乱序问题watermark可以设置延迟时间 2、allowedLateness允许数据的最大延迟时间 3、使用侧输出流 sideOutputLateData
13、如何确定Flink任务的合理并行度在Flink中任务的并行度和消费Kafka分区数据之间有什么关系
最优并行度计算开发完成之后先压测任务并行度给10以下测试单个并行度的处理上限然后 总 QPS / 单个并行度的处理上限 并行度。 最好根据高峰期的 QPS 压测 并行度 * 1.2 留有一些富裕资源。
source 端并行度如果上游数据源是kafka那么并行度与kafka分区保持一致。如果一致的情况下还是消费不过来反压考虑kafka扩大分区并且flink的并行度与分区数保持一致。flink的一个并行度可以处理多个分区数据如果并行度多于分区数那么就会造成并行度空闲浪费资源。
transform 端并行度keyby 之前的算子一般都是跟 source 保持一致。keyby 之后如果并发较大建议设置并行度为 2的整数次幂 。
sink 端并行度 Sink 端是数据流向下游的地方可以根据 Sink 端的数据量 及 下游的服务抗压能力 进行评估。如果 Sink 端是 Kafka可以设为 Kafka 对应 Topic 的分区数。Sink 端的数据量小比较常见的就是监控告警的场景并行度可以设置的小一些。如果 Sink 端的数据量非常大那么在 Sink 到下游的存储中间件的时候就需要提高并行度。
另外 Sink 端要与下游的服务进行交互并行度还得根据下游的服务抗压能力来设置如果在 Flink Sink 这端的数据量过大的话且 Sink 处并行度也设置的很大 但下游的服务完全撑不住这么大的并发写入可能会造成下游服务直接被写挂所以最终还是要在 Sink 处的并行度做一定的权衡。
14、请讨论Flink的状态管理机制包括状态的类型和如何使用。
Flink有两种基本类型的状态托管状态Managed State和原生状态Raw State。从名称中也能读出两者的区别Managed State是由Flink管理的Flink帮忙存储、恢复和优化Raw State是开发者自己管理的需要自己序列化。
Managed StateRaw State状态管理方式Flink Runtime托管自动存储、自动恢复、自动伸缩用户自己管理状态数据结构Flink提供的常用数据结构如ListState、MapState等字节数组byte[]使用场景绝大多数Flink算子用户自己定义
raw state基本不用所以主要介绍managed state。 Flink状态只能在rich function中使用要通过RuntimeContext进行访问managed state主要具有三种状态
Keyed StateOperator StateBroadcast State(1.5版本之后特殊的Operator State)
keyed state 主要是在key stream上保存的状态每一个key都会有对应的一个state支持的类型有valueState 保存一个可以更新和检索的值通过update进行更新通过value进行获取。listState 保存一个元素列表通过 IterableT get()获取通过add或者addAll追加元素通过update进行更新覆盖。reducingState 保存一个单值添加到状态的所有值的聚合接口与 ListState 类似。AggregatingStateIN, OUT 跟reducingState很像但是in和out的类型可以不一致。mapStateUK, UV 维护一个map映射关系通过put或者putAll添加映射通过get或者entries获取跟map的接口一致。
operator state 需要使用operator State时我们可以通过实现checkpointedFunction接口。
这个接口主要时提供了访问non-keyed state的方法主要是需要实现以下两种方法
void snapshotState(FunctionSnapshotContext context) throws Exception;void initializeState(FunctionInitializationContext context) throws Exception;
operator state 支持的类型listState、unionListState、BroadcastState
getUnionListState(descriptor)会使用union redistribution算法 而getListState(descriptor)则简单的是使用even-split redistribution算法。
Even-split redistribution: 每个算子都保存一个列表形式的状态集合整个状态由所有的列表拼接而成。当作业恢复或重新分配的时候整个状态会按照算子的并发度进行均匀分配。 比如说算子 A 的并发读为 1包含两个元素 element1 和 element2当并发读增加为 2 时element1 会被分到并发 0 上element2 则会被分到并发 1 上。
Union redistribution: 每个算子保存一个列表形式的状态集合。整个状态由所有的列表拼接而成。当作业恢复或重新分配时每个算子都将获得所有的状态数据。 Do not use this feature if your list may have high cardinality. Checkpoint metadata will store an offset to each list entry, which could lead to RPC framesize or out-of-memory errors.简单来说就是如果数据量基数过大那么不要用这种方法因为checkpoint的meta数据可能会导致OOM
并行度变化时 注意keyed-state 不能在 open 方法中访问、更新 state这是不行的因为 open 方法在执行时还没有到正式的数据处理环节上下文中是没有 key 的
15、Flink的广播流是什么它有什么用途
在Flink中广播流是一种特殊的数据流类型用于将一个数据流广播到所有并行任务中以供每个任务共享和使用。在流上调用DataStream.broadcast()方法并传入MapStateDescriptor作为状态描述符就可以将它转化为广播流BroadcastStream。该方法的源码如下注意MapStateDescriptor可以有多个。
public BroadcastStreamT broadcast(final MapStateDescriptor?, ?... broadcastStateDescriptors) {Preconditions.checkNotNull(broadcastStateDescriptors);final DataStreamT broadcastStream setConnectionType(new BroadcastPartitioner());return new BroadcastStream(environment, broadcastStream, broadcastStateDescriptors);
}
如果一个算子有多项任务而它的每项任务状态又都相同那么这种特殊情况最适合应用广播状态比如某些规则或者维表。
在传入的 BroadcastProcessFunction 或 KeyedBroadcastProcessFunction 中我们需要实现两个方法。processBroadcastElement() 方法负责处理广播流中的元素processElement() 负责处理非广播流中的元素。 两个子类型定义如下
public abstract class BroadcastProcessFunctionIN1, IN2, OUT extends BaseBroadcastProcessFunction {public abstract void processElement(IN1 value, ReadOnlyContext ctx, CollectorOUT out) throws Exception;public abstract void processBroadcastElement(IN2 value, Context ctx, CollectorOUT out) throws Exception;
}public abstract class KeyedBroadcastProcessFunctionKS, IN1, IN2, OUT {public abstract void processElement(IN1 value, ReadOnlyContext ctx, CollectorOUT out) throws Exception;public abstract void processBroadcastElement(IN2 value, Context ctx, CollectorOUT out) throws Exception;public void onTimer(long timestamp, OnTimerContext ctx, CollectorOUT out) throws Exception;
}
这两个方法的区别在于对 broadcast state 的访问权限不同。在处理广播流元素这端是具有读写权限的而对于处理非广播流元素这端是只读的。 这样做的原因是Flink 中是不存在跨 task 通讯的。所以为了保证 broadcast state 在所有的并发实例中是一致的我们在处理广播流元素的时候给予写权限在所有的 task 中均可以看到这些元素并且要求对这些元素处理是一致的 那么最终所有 task 得到的 broadcast state 是一致的。注processBroadcastElement() 的实现必须在所有的并发实例中具有确定性的结果。注册一个定时器只能在 KeyedBroadcastProcessFunction 的 processElement() 方法中进行。 在 processBroadcastElement() 方法中不能注册定时器因为广播的元素中并没有关联的 key。
重要注意事项 没有跨 task 通讯如上所述这就是为什么只有在 (Keyed)-BroadcastProcessFunction 中处理广播流元素的方法里可以更改 broadcast state 的内容。 同时用户需要保证所有 task 对于 broadcast state 的处理方式是一致的否则会造成不同 task 读取 broadcast state 时内容不一致的情况最终导致结果不一致。 broadcast state 在不同的 task 的事件顺序可能是不同的虽然广播流中元素的过程能够保证所有的下游 task 全部能够收到但在不同 task 中元素的到达顺序可能不同。 所以 broadcast state 的更新不能依赖于流中元素到达的顺序。 所有的 task 均会对 broadcast state 进行 checkpoint虽然所有 task 中的 broadcast state 是一致的但当 checkpoint 来临时所有 task 均会对 broadcast state 做 checkpoint。 这个设计是为了防止在作业恢复后读文件造成的文件热点。当然这种方式会造成 checkpoint 一定程度的写放大放大倍数为 p并行度。Flink 会保证在恢复状态/改变并发的时候数据没有重复且没有缺失。 在作业恢复时如果与之前具有相同或更小的并发度所有的 task 读取之前已经 checkpoint 过的 state。在增大并发的情况下task 会读取本身的 state多出来的并发p_new - p_old会使用轮询调度算法读取之前 task 的 state。 不使用 RocksDB state backend: broadcast state 在运行时保存在内存中需要保证内存充足。这一特性同样适用于所有其他 Operator State。
16、Flink是否支持JobMaster的高可用性HA其原理是什么
高可用一般概念是指在任何时候都有一个领导者 jobManager如果领导者出现故障则有多个备用JM来接管领导。保证不存在单点故障。可以通过zookeeper和Kubernetes提供以下高可用服务
leader选举 从n个候选者中选出一个leader服务发现检索当前leader的地址状态持久化继承程序恢复作业所需的持久化状态JobGraph、用户代码jar、已完成的检查点
17、在不重启Flink的前提下如何动态修改Flink的配置
1通过广播流去读取最新配置文件然后将最新的配置广播出去与主流进行connect。
2通过维表关联比如lookupjoin去读取mysql或者redis、tidb等存储的配置信息然后创建动态表或者其他API我也不知道还有啥。。。进行更新最新配置。
如果还有啥方法欢迎大佬们在评论区补充我就想到这些了。。
18、在Flink中如何实现实时的Top N处理
直接看代码。。。
public class WaterMarkWaitingDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSensorReading sourceData env.addSource(new SensorSource()) // 随机写的一个数据源.assignTimestampsAndWatermarks(WatermarkStrategy.SensorReadingforBoundedOutOfOrderness(Duration.ofMillis(3000)).withTimestampAssigner((r, ts) - r.timestamp));//采用事件时间DataStreamTuple3String, Integer, Long data sourceData.keyBy(r - r.id).window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5))) // 前一个统计数量后一个加上end窗口时间.aggregate(new MyAggregateFunction(), new MyProcessFunction());data.keyBy(r - r.f2).process(new TopN(3)).print();env.execute();}public static class MyAggregateFunction implements AggregateFunctionSensorReading, Integer, Integer {Overridepublic Integer createAccumulator() {return 0;}Overridepublic Integer add(SensorReading sensorReading, Integer integer) {return integer 1;}Overridepublic Integer getResult(Integer integer) {return integer;}Overridepublic Integer merge(Integer integer, Integer acc1) {return null;}}public static class MyProcessFunction extends ProcessWindowFunctionInteger, Tuple3String, Integer, Long, String, TimeWindow {Overridepublic void process(String s, Context context, java.lang.IterableInteger iterable, CollectorTuple3String, Integer, Long collector) throws Exception {Integer cnt iterable.iterator().next();// 迭代器里面只有一个元素因为前面聚合函数每个key其实只会有一个聚合后的最终value可以仔细想想collector.collect(Tuple3.of(s, cnt, context.window().getEnd()));}}public static class TopN extends KeyedProcessFunctionLong, Tuple3String, Integer, Long, String {private MapLong, ListTuple3String, Integer, Long data;// 个人感觉用mapstate更好点。。。private int n;public TopN(int n) {this.n n;data new HashMap();}Overridepublic void processElement(Tuple3String, Integer, Long value, Context context, CollectorString collector) throws Exception {Long end value.f2;ListTuple3String, Integer, Long tmp new ArrayList();//这里可以考虑剪枝只保留n个元素就行if (data.containsKey(end)) {tmp data.get(end);tmp.add(value);} else {tmp.add(value);data.put(end, tmp);}context.timerService().registerEventTimeTimer(end 1); // end窗口1表示一定触发窗口了}Overridepublic void onTimer(long timestamp, OnTimerContext ctx, CollectorString out) throws Exception {super.onTimer(timestamp, ctx, out);ListTuple3String, Integer, Long tmp data.get(ctx.getCurrentKey());tmp.sort(new ComparatorTuple3String, Integer, Long() {Overridepublic int compare(Tuple3String, Integer, Long o1, Tuple3String, Integer, Long o2) {return o2.f1 - o1.f1;}});StringBuilder sb new StringBuilder();sb.append(\n);for (int i0;iMath.min(n, data.size()); i) {Tuple3String, Integer, Long t tmp.get(i);sb.append(top (i1) \n);sb.append(id t.f0 \n);sb.append(cnt t.f1 \n);sb.append(window end t.f2 \n);sb.append(\n);}tmp.clear();out.collect(sb.toString());}}
}19、Flink SQL是如何进行查询解析和优化的
等我出个新的博客文章。。太多了。。。
20、请解释一下Flink的流批一体架构。
Flink作为流批一体化的框架其中流式处理是使用DataStream而批处理则是使用DataSet。后面的版本已经合并只需要维护一套DataStream API。其中由以下几个核心组件。 主要执行批任务还是流任务主要根据数据源是有届还是无界判断。
21、Flink和Spark Streaming在流处理方面有什么不同在什么情况下应该选择使用Flink与其他流处理框架相比Flink有哪些优点为什么选择使用Flink而不是其他微批处理框架你考虑过哪些因素
等我学完spark一定补上。。。。
22、请介绍一下Flink的复杂事件处理CEP库并给出使用场景的例子。
23、使用Flink Client消费Kafka数据和使用Flink Connector消费有什么不同
