flink20问-3
1、Flink 的作业调度机制是如何设计的？如何优化作业的调度效率？
作业调度机制主要依赖于内部的 jobmanager 和 taskmanager 进行协同工作，jobmanager 负责协调、调度和监控作业，而 taskmanager 则执行实际的作业任务，作业调度过程包含：
a、作业提交：用户通过 flink 提交作业，作业被解析并转换成 jobgraph
b、jobmanager 接受作业：jobgraph 被提交到 jobmanager，jobmanager 将其转换成 executiongraph
c、资源分配：jobmanager 向资源管理框架如 yarn，请求 taskmanager，确保有足够的资源来执行作业
d、任务分配：在资源充足的情况下，jobmanager 会将 executiongraph 切分成多个并行子任务，并将这些任务分配给 taskmanager
e、任务执行：taskmanager 接受到任务后，开始执行对应计算，并将结果发送回 jobmanager 进行监控和调度
优化 flink 作业的调度效率，可以考虑以下几个方面：
a、资源管理和分配：确保有足够的 TaskManager 资源，并通过合理配置资源限制，避免资源的浪费；
b、并行度优化：根据作业的实际需求和资源情况，动态调度作业的并行度，以便充分利用系统资源
c、优先级和作业依赖关系：对于复杂的多作业系统，设置作业的优先级和了解作业的依赖关系，有助于提升整体调度效率；
d、容错机制制造设计：使用 flink 的 checkpoint 和 savepoint 机制，提升作业的容错能力，避免因任务失败导致的重新调度开销
e、数据本地化：尽量将任务调度到数据所在的节点，减少数据传输带来的开销
f、优化网络带宽：通过优化数据传输和减少 shuffle 操作，提升网络带宽的利用效率
jobgraph：用户提交作业的逻辑概述，包含作业各个算子的拓扑结构
executiongraph：将作业逻辑转换为可执行的并行任务，包含详细的执行计划和任务分配信息
slot：taskmanager 上的可用资源单位，每个 slot 可以运行一个字任务；
slot pool：jobmanager 中管理全部 slot 的机制，通过动态分配 slot，提高资源利用率
flink 提供了多种调度策略：如 lazy from sources 调度策略，允许从作业源头开始，逐步调度后续节点，最大限度利用数据流特点。
2、在 Flink 中，如何处理反压问题？有哪些常见的反压解决策略？
flink 中，反压是指流处理过程中，下游的任务处理速度跟不上上游的任务，从而造成数据积压，要处理反压问题，可以采取以下几种常见的策略：
a、优化并行度
b、增加资源：cpu、内存
c、调整缓冲区大小，增加任务之间网络缓冲区的大小，通过 taskmanager.network.memory.fraction、taskmanager.network.memory.min、taskmanager.network.memory.max 等参数可以有效缓解缓冲区太小导致频繁溢出最终导致反压的问题
d、数据预处理：在数据进入 flink 流之前，进行预处理，减少进入 flink 的数据量
e、调整 checkpoint 配置：合理配置 checkpoint 的频率和超时，避免由于频繁的 checkpoint 导致的反压
f、使用分布式缓存如 redis 来减轻数据库的压力
3、Flink 中的连接操作 Join 是如何实现的？如何优化 Join 的性能？
join 操作通过两条数据流根据某个公共键进行配对，flink 提供多种 join 策略，包括窗口 join、间隔 join 和 内部 join；
窗口 join：窗口 join 在设定的窗口事件内对两条流进行连接，适用于事件时间模式； 例如两个流的订单和支付信息，通过时间窗口可以使得某个时间范围的订单和支付进行匹配
stream1.join(stream2)
.where(<KeySelector>)
.equalTo(<KeySelector>)
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.apply(new JoinFunction<...>());
间隔 join：允许在特定时间间隔内匹配两个流中的元素； 适用于点击行为和广告曝光匹配等场景
stream1.keyBy(...)
.intervalJoin(stream2.keyBy(...))
.between(Time.seconds(-5), Time.seconds(5))
.process(new ProcessJoinFunction<...>());
普通内连接等可以基于字段进 join；
stream1.join(stream2)
.where(<KeySelector>)
.equalTo(<KeySelector>)
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
.apply(new JoinFunction<...>());
优化 join：
a、选择合适的窗口类型：不同窗口类型对性能有显著影响，例如滚动窗口、滑动窗口等；
b、过滤无用数据：在 join 之前对数据进行预过滤，以减少不必要的数据传输和内存消耗
c、优化分区：确保 join 操作的数据流按照同一键进行分区，以减少数据交换和分布式计算的开销
d、资源管理：分配足够的资源以支持 join 操作
4、Flink 如何支持基于窗口的流处理？如何选择合适的窗口类型？
滚动窗口：大小固定不重叠，适用于对数据进行周期性汇总，比如每小时的访问量统计；
滑动窗口：可以设定窗口大小和滑动步长，适用于需要高频计算的场景，比如实时监控系统中的指标计算；
会话窗口：用于处理用户活动的数据流，适用于分析用户行为，比如电商网站的用户会话时长统计
// 创建环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 设定数据源
DataStream<String> input = env.socketTextStream("localhost", 9999);
// 滚动窗口示例
DataStream<Int> tumblingWindowResult = input
.map(new MapFunction<String, Int>() {
public Int map(String value) { return Integer.parseInt(value); }
})
.timeWindowAll(Time.minutes(5))
.sum(0);
// 滑动窗口示例
DataStream<Int> slidingWindowResult = input
.map(new MapFunction<String, Int>() {
public Int map(String value) { return Integer.parseInt(value); }
})
.timeWindowAll(Time.minutes(10), Time.minutes(5))
.sum(0);
// 开始执行
env.execute("Flink Window Example");
5、Flink 的 GlobalWindow 和 TimeWindow 有什么区别？如何选择使用？
globalwindow 这个窗口没有设定的时间范围，它会一直积累元素，直到触发某种条件，比如通过 trigger 来决定何时计算窗口中的数据，通常与自定义触发器配合使用；
简单来说，就是 timewindow 自带默认的出发器，而 globalwindow 需要自定义触发器；适用于电商平台上的购物车数据汇总，如用户点击结算按钮来触发计算；
6、Flink 的 Changelog 是如何设计的？它在状态恢复中起到什么作用？
changelog 是一种通过记录状态变化来便于状态管理和恢复的机制，在分布式计算中，状态是指支持流式计算的所有数据，通过记录状态的变化日志，可以提高系统在故障后的恢复速度和准确性；
具体作用如下：
a、更快的恢复速度：changlog 记录了状态的增量变化，恢复时可以先加载最新的快照再应用 changlog 中记录的增量变化，减少了恢复所需的时间；
b、提高准确性：changlog 可以确保所有状态变化都被记录下来，在恢复过程中可以准确还原状态，避免数据丢失或不一致；
c、减少存储空间：相比于频繁保存完整的状态快照，记录增量变化更为高效，这样既节省了存储空间，又减轻了网络传输的负担；
flink 定期对状态进行快照，快照包含当前所有状态的完整复本。快照策略有：时间间隔、事件驱动；
在增量快照模式下，flink 只记录自上一次成功快照以来的状态变化，对稳定运行的系统，这能大大减少存储需求和快照时间；
flink 支持精确一次的语义，在这种语义下，通过 changlog 可以确保即使在故障后恢复也能确保每条记录被处理且仅被处理一次；
当任务失败时，flink 通过最近的快照和 changlog 中的记录进行恢复，先应用快照，再应用 changelog 里的增量变化，恢复到最新状态；
7、Flink 的批处理和流处理在底层实现上有什么不同？如何统一批处理和流处理？
a、数据视图：批处理视图为有界数据集，而流处理视图为无界数据流；
b、计算模型：批处理一般采用 mapreduce 模型，流处理则基于事件驱动和持续计算；
c、执行计划：批处理可以优化执行计划，因为数据集是固定的；流处理需要适应数据的不断到达，不能提前预知和优化；
d、一致性保障：批处理通过重试机制，确保任务最终一致性；流处理通过时间窗口和 checkpoint 机制，实现实时一致性；
8、Flink 的容错机制是如何通过两阶段提交实现的？
第一阶段：在这一步，flink 的各种操作，如 source、function、sink 都会生成一个全局一致的快照 checkpoint，每个任务将状态保存到分布式存储如 hdfs 中，但并不会立即提交；
第二阶段：当 flink 确定所有任务都成功完成了第一阶段的操作后，才会统一执行提交操作，所有参与者，共同完成最终的状态提交，确保数据的一致性；
9、在 Flink 中，如何进行流数据的去重处理？有哪些常见的去重方法？
a、基于状态编程：
通过 flink 的状态管理来进行去重，可以使用 KeyedState 来管理已经处理过的记录；例如，常见的方式是用 ValueState 保存已经处理的数据的标记，如果新来的数据在状态中已经存在，则表示是重复数据，可以丢弃；否则，就更新状态并处理数据，这种方式适用于数据量较小的场景，因为保存历史记录可能占用较多内存；
DataStream<MyEvent> stream = ...;
stream.keyBy(MyEvent::getKey)
.process(new KeyedProcessFunction<String, MyEvent, MyEvent>() {
private static final long serialVersionUID = 1L;
private ValueState<Boolean> seen;
@Override
public void open(Configuration parameters) throws Exception {
seen = getRuntimeContext().getState(new ValueStateDescriptor<>("seen", Types.BOOLEAN));
}
@Override
public void processElement(MyEvent event, Context ctx, Collector<MyEvent> out) throws Exception {
if (seen.value() == null) {
seen.update(true);
out.collect(event);
}
}
});
b、使用布隆过滤器
布隆过滤器通过多个哈希函数来记录元素的存在情况，具有极小的误判率。用布隆过滤器进行去重处理在内存占用上更有优势，适合处理大规模数据；
DataStream<MyEvent> stream = ...;
stream.keyBy(MyEvent::getKey)
.map(new RichMapFunction<MyEvent, MyEvent>() {
private transient ValueState<BloomFilter> bloomFilterState;
@Override
public void open(Configuration parameters) throws Exception {
ValueStateDescriptor<BloomFilter> descriptor = new ValueStateDescriptor<>(
"bloomFilter",
BloomFilter.class
);
bloomFilterState = getRuntimeContext().getState(descriptor);
}
@Override
public MyEvent map(MyEvent event) throws Exception {
BloomFilter bloomFilter = bloomFilterState.value();
if (bloomFilter == null) {
bloomFilter = new BloomFilter(1000000, 0.01);
}
if (!bloomFilter.mightContain(event.getUniqueId())) {
bloomFilter.put(event.getUniqueId());
bloomFilterState.update(bloomFilter);
return event;
}
return null;
}
})
.filter(Objects::nonNull);
c、ttl 策略结合状态编程
在一些实时流处理场景下，旧数据过一段时间后就没必要再去重，这时候可以结合 ttl 策略管理状态，控制状态的生命周期，定期清理不活跃的状态；
DataStream<MyEvent> stream = ...;
stream.keyBy(MyEvent::getKey)
.window(TumblingProcessingTimeWindows.of(Time.minutes(10)))
.process(new ProcessWindowFunction<MyEvent, MyEvent, String, TimeWindow>() {
private transient ValueState<Boolean> seen;
@Override
public void open(Configuration parameters) throws Exception {
StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(Time.minutes(30))
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
.build();
ValueStateDescriptor<Boolean> descriptor = new ValueStateDescriptor<>(
"seen", Types.BOOLEAN
);
descriptor.enableTimeToLive(ttlConfig);
seen = getRuntimeContext().getState(descriptor);
}
@Override
public void process(String key, Context context, Iterable<MyEvent> elements, Collector<MyEvent> out) throws Exception {
if (seen.value() == null) {
seen.update(true);
for (MyEvent event : elements) {
out.collect(event);
}
}
}
});
10、Flink 的迭代处理是如何实现的？有哪些典型的应用场景？
通过两个主要的算子实现：Iteration 和 DeltaIteration；它们允许数据在计算图中进行循环迭代，以便在满足某些条件时产生新的数据；flink 中的迭代处理适用于多个典型的应用场景，例如：迭代训练机器学习算法、图处理、网络分析等；
a、Iteration：flink 最基础的迭代算子，它允许在逻辑数据流中定义一个循环，直到满足某种退出条件，例如可以用 Iteration 算子来进行数据聚合，多次迭代直至结果收敛；
b、DeltaIteration：不同于 Iteration，DeltaIteration 仅处理那些在每一轮迭代中发生变化的数据，即“增量”数据，因为它只处理变化的部分，所以可以减少数据处理量，并加快收敛速度；
11、Flink 的 Checkpoint 与 Savepoint 有什么区别？在什么场景下使用 Savepoint？
checkpoint 是 flink 内部自动创建的，用于容错机制，savepoint 是用户手动触发的，用于特定场景下的状态保存；如作业升级、迁移或暂停和恢复等操作，允许用户在这些场景中手动管理和恢复状态；
12、在 Flink 中，如何进行时间窗口的聚合操作？有哪些常见的聚合函数？
创建流环境 Stream Execution Environment - 导入数据源 Stream Source 如 Kafka、文件等 - 指定时间特性，如处理时间、事件事件 - 定义时间窗口 - 应用聚合函数，聚合函数：sum、max、min、reduce（自定义的聚合逻辑，可以实现更复杂的聚合操作）
// 1. 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 2. 从 Kafka 获取数据
Properties properties = new Properties();
// 设置必要的 Kafka 配置属性比如 bootstrap.servers 等
DataStream<String> stream = env.addSource(new FlinkKafkaConsumer<>("my-topic", new SimpleStringSchema(), properties));
// 3. 定义数据类型及其时间特性
DataStream<Tuple2<String, Integer>> parsedStream = stream
.map(line -> { // 假设每行都是 'key,value' 格式
String[] parts = line.split(",");
return new Tuple2<>(parts[0], Integer.parseInt(parts[1]));
})
.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<Tuple2<String, Integer>>() {
@Override
public long extractAscendingTimestamp(Tuple2<String, Integer> element) {
return System.currentTimeMillis(); // 处理时间作为时间戳
}
});
// 4. 定义窗口和聚合操作
DataStream<Tuple2<String, Integer>> result = parsedStream
.keyBy(0) // 按照 Tuple2 的第一个字段（key）分组
.timeWindow(Time.minutes(5)) // 每5分钟一个翻滚窗口
.sum(1); // 对 Tuple2 的第二个字段（value）求和
// 5. 输出或者进一步处理
result.print();
try {
env.execute("Windowed Aggregation Example");
} catch (Exception e) {
e.printStackTrace();
}
13、Flink 的水印策略是如何设计的？如何选择合适的水印生成策略？
Flink 的水印策略是用来处理无序到达的事件流的，它提供了一种机制使得窗口操作在处理无序事件流时可以正确计算；
a、水印策略设计：Flink 的水印策略核心是为事件打上时间标记，并周期性地产生水印，在事件流的处理中，每一条事件都会附带一个时间戳，根据这些时间戳生成水印；水印本质上是特殊的标记，它表示在某个时刻之前的所有事件都已经到达；
b、常见的水印生成策略有如下几种：
固定延迟：适用于事件到达延迟较为固定的场景，每隔固定的时间产生一次水印；
自适应延迟：适用于事件到达延迟不确定的场景，根据历史数据动态调整水印生成的时间；
间隔水印：适用于明确有特定事件标志需要生成水印的场景，例如来自特定标记的用户事件
14、Flink 的动态分区机制是如何实现的？它对大规模流式数据处理有何优化作用？
动态分区机制是基于数据流的特性，对数据流中的记录进行自动分区，实现数据在不同节点上的负载均衡，从而优化数据处理性能；动态分区允许在运行时感觉某些条件（例如字段值、时间戳等）调整数据的路由策略，使得数据能够更均匀地分布在各个计算节点上，避免数据倾斜问题，提升系统的吞吐量和处理能力；
在大规模流式数据处理过程中，Flink 的动态分区机制能够：
a、均衡负载，防止某些节点过载，导致整体性能下降；
b、提高系统的并行度和扩展性；
c、减少数据倾斜，确保数据处理的效率和稳定性；
d、实现更灵活的数据路由，适应不同的数据流特性和处理需求；
实现原理：flink 的动态分区机制依赖于底层的 datastream api，通过 .keyBy() 方法，可以对数据流进行分区，这个方法会感觉提供的分区键动态计算每条记录的目标分区，在对分区键进行哈希计算后，flink 会将处理任务分配给相应的并行实例；
分区策略：轮询分区、随机分区、哈希分区、广播分区；
数据倾斜问题的解决：预分区（在数据进入 flink 集群之前，进行初步的分区）、负载均衡、联合建（采用多个字段组成分区键，降低数据倾斜的可能性）
优化性能：并行化处理、资源节省、灵活性
15、Flink 的任务 Slot 和并行度如何影响作业的执行效率？
任务 slot：每一个 slot 是一个并行单位，代表一个 taskmanager 可以运行的任务数量，分配足够的 slot 可以提升任务的并行执行能力，从而提高整个作业的执行效率；
并行度：并行度决定了每个操作符的多少个子任务能够同时执行，较高的并行度能够使更多的子任务同时进行，从而加快数据处理速度；
16、Flink 如何在任务失败时进行自动恢复？自动恢复的机制是什么？
checkpoint + savepoint
17、在 Flink 中，如何通过 State TTL 管理状态的生命周期？
state ttl 时用来管理状态的生命周期，防止状态数据无限增长并导致内存占用过高，通过 state ttl，可以设定状态的过期时间，当状态超过设置的 ttl 时间时，flink 会将其自动清理，配置 state ttl 主要通过：
a、创建 StateDescriptor 包括 ValueStateDescriptor、ListStateDescriptor 等；
b、设置 ttl 配置（StateTtlConfig）
c、通过 StateDescriptor 使用 TTL 配置；
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
public class StateTtlExample {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 设置 TTL 配置
StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(org.apache.flink.api.common.time.Time.seconds(60)) // 设置状态的TTL为60秒
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) // 在状态创建和写入时更新TTL
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) // 从不返回过期状态
.build();
ValueStateDescriptor<Long> stateDescriptor =
new ValueStateDescriptor<>("exampleState", Long.class);
stateDescriptor.enableTimeToLive(ttlConfig);
// 添加 KeyedProcessFunction
env.socketTextStream("localhost", 9999)
.keyBy(value -> value)
.process(new KeyedProcessFunction<String, String, String>() {
private transient ValueState<Long> state;
@Override
public void open(Configuration parameters) throws Exception {
state = getRuntimeContext().getState(stateDescriptor);
}
@Override
public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
Long currentState = state.value();
if (currentState == null) {
currentState = 0L;
}
currentState += 1;
state.update(currentState);
out.collect("Key: " + value + ", Count: " + currentState);
}
});
env.execute("State TTL Example");
}
}
18、Flink 如何实现容错保证 Exactly-Once 语义？底层的原理是什么？
通过 Checkpoints 机制和状态后端来完成，Checkpoint 定期为作业的中间状态创建一致性快照，确保即使发生故障，也能恢复到一致的状态
19、在 Flink 中，如何通过重平衡操作优化数据分布？
重平衡操作可以将数据重新平衡分布到所有并行子任务中，以优化数据处理的负载均衡，具体操作是使用 rebalance() 方法，它会随机地将数据重新分布到下游的所有并行任务，从而避免某些任务因为数据过多而成为处理瓶颈；
DataStream<MyType> stream = ...;
DataStream<MyType> rebalancedStream = stream.rebalance();
会将数据流的每一条记录随机均匀分布到下游所有的并行任务中，这种机制确保了每个下游任务接收到的数据量大致相同，从而优化了并行处理的性能；
与 shuffle 的区别：重平衡是随机均匀分布数据，适用于大多数情况下的数据均衡；
shuffle：带有一定的组分配机制，适用于需要特定数据分布的场景，比如某些迭代算法；
20、Flink 中的增量 Checkpoint 是如何实现的？它的优势和应用场景是什么？
a、首次 checkpoint 全量保存所有状态数据
b、在进行下一次 checkpoint 之前，flink 持续监控状体啊的变化，并积累这些变化；
c、在后续 checkpoint 中，只保存自上次 checkpoint 以来的状态增量；
d、在恢复时，从最后一个全量 checkpoint 开始应用所有后续增量，重建当前状态；
优势：
a、节省空间和时间
b、提升性能
c、适用于大规模、状态变化频繁的任务