从零开始学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的执行模式经历了以下几个关键阶段：

1. 早期版本： Flink最初专注于流处理，但提供了对批处理的支持
2. Flink 1.12： 引入了全新的批处理执行模式（BATCH模式）
3. Flink 1.14： 增强了批处理模式的性能和功能
4. Flink 1.20.1： 进一步优化了批流一体架构，改进了执行模式的自动选择机制

二、Execution Mode的技术原理

1. 两种执行模式的核心区别

虽然Flink使用相同的API和代码结构，但BATCH和STREAMING模式在内部执行方式上存在显著差异：

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文本数据的性能对比：

七、Flink 1.20.1中的执行模式改进

1. 新特性和优化

Flink 1.20.1在执行模式方面带来了多项改进：

• 更智能的AUTOMATIC模式： 改进了自动模式的选择逻辑，支持更复杂的数据源组合
• 批处理模式性能提升： 进一步优化了批处理执行引擎，提升了大数据量处理能力
• API一致性增强： 确保所有算子在不同执行模式下行为一致
• 资源利用率优化： 改进了批处理模式下的资源调度，减少资源浪费

2. 兼容性注意事项

在使用Flink 1.20.1的执行模式时，需要注意以下兼容性问题：

• 某些流处理特有的操作（如CEP）在BATCH模式下可能行为受限
• 窗口操作在BATCH和STREAMING模式下的实现方式不同
• 状态过期机制在两种模式下有细微差别

八、最佳实践

1. 执行模式选择指南

2. 实际应用中的模式切换策略

在实际项目中，可以采用以下策略来管理执行模式：

• 开发环境： 使用AUTOMATIC模式，方便测试不同数据源
• 生产环境： 根据明确的数据流特征选择BATCH或STREAMING模式
• 批处理作业： 明确设置为BATCH模式以获得最佳性能
• 流处理作业： 明确设置为STREAMING模式，确保低延迟

九、总结与展望

Flink的批流一体执行模式是大数据处理领域的一次重要创新，它消除了批处理和流处理之间的界限，为开发者提供了统一、灵活的编程模型。通过Execution Mode的合理选择和配置，我们可以在不同场景下获得最佳的性能表现。

随着Flink 1.20.1的发布，批流一体架构进一步成熟，执行模式的自动选择更加智能，性能优化更加到位。未来，Flink将继续完善其批流一体架构，为大数据处理提供更加强大和灵活的解决方案。

通过本文的学习，相信你已经对Flink的执行模式有了深入的理解。在实际应用中，建议根据具体的数据特征和处理需求，选择合适的执行模式，充分发挥Flink批流一体的优势。

原文来自：http://blog.daimajiangxin.com.cn

源码地址：https://gitee.com/daimajiangxin/flink-learning