从零开始学Flink：流批一体的执行模式
在大数据处理领域，批处理和流处理曾经被视为两种截然不同的范式。然而，随着Apache Flink的出现，这种界限正在逐渐模糊。Flink的一个核心特性是其批流一体的架构设计，允许用户使用统一的API和执行引擎处理有界数据（批处理）和无界数据（流处理）。本文将深入探讨Flink的执行模式（Execution Mode），特别是在Flink 1.20.1版本中对批处理和流处理模式的支持和优化。
一、Flink执行模式概述
1. 执行模式的基本概念
Flink的执行模式决定了作业如何被调度和执行。在Flink 1.12及以后的版本中，引入了统一的流批处理执行模式，主要包括以下三种模式：
STREAMING模式： 传统的流处理执行模式，适用于处理无界数据流
BATCH模式： 专门为有界数据优化的批处理执行模式
AUTOMATIC模式： 自动根据数据源类型选择执行模式
这三种模式的引入使得Flink能够在同一套API上提供最佳的批处理和流处理性能。
2. 执行模式的演进历程
Flink的执行模式经历了以下几个关键阶段：
早期版本： Flink最初专注于流处理，但提供了对批处理的支持
Flink 1.12： 引入了全新的批处理执行模式（BATCH模式）
Flink 1.14： 增强了批处理模式的性能和功能
Flink 1.20.1： 进一步优化了批流一体架构，改进了执行模式的自动选择机制
二、Execution Mode的技术原理
1. 两种执行模式的核心区别
虽然Flink使用相同的API和代码结构，但BATCH和STREAMING模式在内部执行方式上存在显著差异：
特性 STREAMING模式 BATCH模式
调度策略 连续流式调度 批处理调度，类似于MapReduce
资源利用 持续占用资源 任务完成后释放资源
优化技术 流式优化 批处理优化，如查询优化、物化视图
处理延迟 毫秒级延迟 较高延迟，但吞吐量更大
适用场景 实时数据处理 离线数据分析
2. 批流一体的设计理念
Flink的批流一体架构基于以下核心理念：
统一的API： 无论批处理还是流处理，都使用相同的DataStream API
统一的状态管理： 共享相同的状态后端和检查点机制
统一的容错机制： 基于检查点的故障恢复
统一的优化器： 但针对不同执行模式应用不同的优化策略
三、配置和使用Execution Mode
1. 环境准备
首先，确保你已经设置了正确的依赖：
dependencies {
// Flink核心依赖
implementation 'org.apache.flink:flink_core:1.20.1'
implementation 'org.apache.flink:flink-streaming-java:1.20.1'
implementation 'org.apache.flink:flink-clients:1.20.1'
implementation 'org.apache.flink:flink-connector-files:1.20.1'
implementation 'org.apache.flink:flink-connector-kafka:3.4.0-1.20'
}
2. 在代码中设置执行模式
在Flink 1.20.1中，可以通过以下方式设置执行模式：
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class ExecutionModeExample {
public static void main(String[] args) throws Exception {
// 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 设置执行模式为BATCH
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
// 或者设置为STREAMING
// env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
// 或者设置为AUTOMATIC（根据数据源自动选择）
// env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
// 后续代码...
}
}
3. 通过命令行参数设置
也可以通过命令行参数覆盖代码中的设置：
bin/flink run -Dexecution.runtime-mode=BATCH -c com.example.ExecutionModeExample your-jar-file.jar
四、BATCH模式与STREAMING模式实践
1. 批处理模式示例
以下是使用BATCH模式处理文件数据的完整示例：
package com.cn.daimajiangxin.flink;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.StreamFormat;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import java.time.Duration;
import java.util.Arrays;
public class BatchWordCount {
public static void main(String[] args) throws Exception {
// 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 明确设置为批处理模式
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
// 从文件读取数据（有界数据源）
String inputPath = "path\\flink-learning\\data\\input.txt";
// 1. 创建文件源构建器
Path filePath = new Path(inputPath);
// 2. 配置文件读取格式
StreamFormat<String> format =new TextLineInputFormat("UTF-8");
// 3. 构建 FileSource
FileSource<String> fileSource = FileSource
.forRecordStreamFormat(format, filePath)
.build();
// 4. 添加 Watermark 策略（批处理中可使用默认策略）
WatermarkStrategy<String> watermarkStrategy = WatermarkStrategy
.<String>forMonotonousTimestamps()
.withIdleness(Duration.ofSeconds(10));
DataStream<String> text = env.fromSource(fileSource,watermarkStrategy,"FileSource");
// 数据处理逻辑
DataStream<Tuple2<String, Integer>> counts = text
.flatMap(new Tokenizer())
.keyBy(value -> value.f0)
.sum(1);
// 输出结果
counts.print();
// 执行作业
env.execute("Batch Word Count");
}
public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
private static final long serialVersionUID = 1L;
@Override
public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
// 分词并为每个单词生成(word, 1)的元组
Arrays.stream(value.toLowerCase().split("\\W+"))
.filter(word -> word.length() > 0)
.forEach(word -> out.collect(new Tuple2<>(word, 1)));
}
}
}
2. 流处理模式示例
以下是使用STREAMING模式处理Kafka数据流的示例：
package com.cn.daimajiangxin.flink;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import java.time.Duration;
import java.util.Arrays;
public class StreamingWordCount {
public static void main(String[] args) throws Exception {
// 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 明确设置为流处理模式
env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
// 启用检查点
env.enableCheckpointing(5000);
// 创建Kafka源（无界数据源）
KafkaSource<String> source = KafkaSource.<String>
builder()
.setBootstrapServers("localhost:9092")
.setTopics("word-count-topic")
.setGroupId("flink-group")
.setStartingOffsets(OffsetsInitializer.earliest())
.setValueOnlyDeserializer(new SimpleStringSchema())
.build();
// 从Kafka读取数据
DataStream<String> text = env.fromSource(
source,
WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(2)),
"Kafka Source"
);
// 数据处理逻辑
DataStream<Tuple2<String, Integer>> counts = text
.flatMap(new Tokenizer())
.keyBy(value -> value.f0)
.sum(1);
// 输出结果
counts.print();
// 执行作业
env.execute("Streaming Word Count");
}
public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
private static final long serialVersionUID = 1L;
@Override
public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
Arrays.stream(value.toLowerCase().split("\\W+"))
.filter(word -> word.length() > 0)
.forEach(word -> out.collect(new Tuple2<>(word, 1)));
}
}
}
五、AUTOMATIC模式的智能选择机制
1. AUTOMATIC模式的工作原理
AUTOMATIC模式是Flink 1.20.1中的一个强大特性，它能够根据作业的数据源类型自动选择最合适的执行模式：
当所有输入源都是有界的（如文件、批量数据库查询），自动选择BATCH模式
当至少有一个输入源是无界的（如Kafka、Socket），自动选择STREAMING模式
// 设置为自动模式
env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
2. 边界情况处理
在某些复杂场景下，AUTOMATIC模式的选择可能不完全符合预期：
混合数据源： 如果作业同时包含有界和无界数据源，将选择STREAMING模式
动态数据源： 对于可能在运行时从有界变为无界的数据源，建议明确指定执行模式
复杂处理拓扑： 对于包含迭代或复杂循环的作业，可能需要手动选择执行模式
六、批处理模式的性能优化
1. 批处理特定的优化
BATCH模式针对有界数据处理提供了多项性能优化：
任务调度优化： 采用更高效的批处理调度策略
内存管理改进： 更积极的数据物化和缓存
网络传输优化： 批量数据传输减少网络开销
计算优化： 使用更适合批处理的算子实现
2. 性能对比示例
使用相同的WordCount逻辑，分别在BATCH和STREAMING模式下处理1GB文本数据的性能对比：
模式 执行时间 CPU使用率 内存消耗
STREAMING 38秒 稳定在70% 2.4GB
BATCH 22秒 峰值95%，完成后释放 1.8GB
七、Flink 1.20.1中的执行模式改进
1. 新特性和优化
Flink 1.20.1在执行模式方面带来了多项改进：
更智能的AUTOMATIC模式： 改进了自动模式的选择逻辑，支持更复杂的数据源组合
批处理模式性能提升： 进一步优化了批处理执行引擎，提升了大数据量处理能力
API一致性增强： 确保所有算子在不同执行模式下行为一致
资源利用率优化： 改进了批处理模式下的资源调度，减少资源浪费
2. 兼容性注意事项
在使用Flink 1.20.1的执行模式时，需要注意以下兼容性问题：
某些流处理特有的操作（如CEP）在BATCH模式下可能行为受限
窗口操作在BATCH和STREAMING模式下的实现方式不同
状态过期机制在两种模式下有细微差别
八、最佳实践
1. 执行模式选择指南
场景 推荐模式 原因
离线数据处理 BATCH 性能更好，资源利用率更高
实时数据处理 STREAMING 低延迟，持续处理能力
ETL作业 BATCH 更适合处理有界数据集
实时分析 STREAMING 满足实时性要求
不确定数据源类型 AUTOMATIC 自动适配不同数据源
2. 实际应用中的模式切换策略
在实际项目中，可以采用以下策略来管理执行模式：
开发环境： 使用AUTOMATIC模式，方便测试不同数据源
生产环境： 根据明确的数据流特征选择BATCH或STREAMING模式
批处理作业： 明确设置为BATCH模式以获得最佳性能
流处理作业： 明确设置为STREAMING模式，确保低延迟
九、总结与展望
Flink的批流一体执行模式是大数据处理领域的一次重要创新，它消除了批处理和流处理之间的界限，为开发者提供了统一、灵活的编程模型。通过Execution Mode的合理选择和配置，我们可以在不同场景下获得最佳的性能表现。
随着Flink 1.20.1的发布，批流一体架构进一步成熟，执行模式的自动选择更加智能，性能优化更加到位。未来，Flink将继续完善其批流一体架构，为大数据处理提供更加强大和灵活的解决方案。
通过本文的学习，相信你已经对Flink的执行模式有了深入的理解。在实际应用中，建议根据具体的数据特征和处理需求，选择合适的执行模式，充分发挥Flink批流一体的优势。