点一下关注吧！！！非常感谢！！持续更新！！！

🚀 AI 篇持续更新中！（长期更新）

AI 炼丹日志-31- 千呼万唤始出来 GPT-5 发布！“快的模型 + 深度思考模型 + 实时路由”，持续打造实用 AI 工具指南！📐🤖

💻 Java 篇正式开启！（300 篇）

目前 2025 年 10 月 13 日更新到：Java-147 深入浅出 MongoDB 分页查询详解：skip() + limit() + sort() 实现高效分页、性能优化与 WriteConcern 写入机制全解析 MyBatis 已完结，Spring 已完结，Nginx 已完结，Tomcat 已完结，分布式服务正在更新！深入浅出助你打牢基础！

📊 大数据板块已完成多项干货更新（300 篇）：

包括 Hadoop、Hive、Kafka、Flink、ClickHouse、Elasticsearch 等二十余项核心组件，覆盖离线+实时数仓全栈！大数据-278 Spark MLib - 基础介绍 机器学习算法 梯度提升树 GBDT 案例 详解

章节内容

上节我们完成了如下的内容：

• Flink Time Watermark

• Java 代码实例测试

Flink 并行度详解

基本概念

Flink 程序由多个算子（Operator）组成，包括数据源（Source）、转换操作（Transformation）和数据输出（Sink）。每个算子在执行时会被拆分为多个并行的任务（Task），这些任务由独立的线程执行。一个算子的并行任务（线程）数量就被称为该算子的并行度（Parallelism）。

并行度的重要性

并行度是 Flink 性能调优的关键参数之一，直接影响：

• 数据处理吞吐量

• 资源利用率

• 任务延迟

• 系统稳定性

并行度设置方式

Flink 提供了多种设置并行度的方式，按优先级从高到低依次为：

1. 算子级别（Operator Level）

dataStream.map(new MyMapFunction()).setParallelism(10);

适用场景：当某个特定算子需要与其他算子采用不同的并行度时使用。

2. 执行环境级别（Execution Environment Level）

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(10);

特点：为整个作业设置默认并行度，会被算子级别的设置覆盖。

3. 客户端级别（Client Level）

flink run -p 10 -c com.example.MyJob myJob.jar

特点：通过命令行提交作业时指定。

4. 系统级别（System Level）

在 flink-conf.yaml 中配置：

parallelism.default: 10

特点：作为全局默认值，当其他级别都未设置时使用。

并行度优化建议

1. 资源匹配：并行度应与可用 slot 数量相匹配

2. 例如集群有 4 个 TaskManager，每个配置 3 个 slot，则总 slot 数为 12

3. 理想情况下作业总并行度不超过总 slot 数

4. 数据倾斜处理：

   // 对可能倾斜的key进行预分散   dataStream.keyBy(key -> key + "#" + ThreadLocalRandom.current().nextInt(10));

4. 典型场景参考值：

5. 简单 ETL 作业：4-8

6. 复杂计算/窗口聚合：8-16

7. 机器学习/图计算：16-32+

动态并行度调整

Flink 1.13+ 支持动态调整并行度：

1. Savepoint 保存状态

2. 修改并行度配置

3. 从 Savepoint 恢复

监控与调优

通过 Flink Web UI 监控：

• 各算子背压情况（Backpressure）

• 各并行子任务处理速率

• 数据倾斜情况

调优原则：从低并行度开始测试，逐步增加直到资源利用率达到 70-80%。

并行度的概念

在 Flink 中，并行度主要决定每个操作符在作业中被分配多少个并发实例来处理数据。操作符的并行实例越多，任务就能够越快完成。通常，Flink 作业中的每个操作符都会以并行实例的形式执行在集群中的不同 TaskManager 上，这样可以充分利用集群的计算资源。

Flink 中的并行度可以分为以下几个层级：

全局并行度（Global Parallelism）

全局并行度是指 Flink 集群默认为所有作业和操作符分配的并行度。在配置文件 flink-conf.yaml 中，你可以通过以下配置来设置 Flink 集群的默认全局并行度：

parallelism.default: 4

这个配置将为每个没有指定并行度的操作符分配默认的 4 个并行实例。如果你没有在代码中或任务提交时明确设置并行度，Flink 将使用这个默认值。

作业并行度（Job-level Parallelism）

在提交 Flink 作业时，你可以为整个作业设置并行度，覆盖全局默认值。例如，在命令行使用 flink run 提交作业时可以通过 -p 参数来设置并行度：

flink run -p 10 your-job.jar

此命令将作业的并行度设置为 10，作业中的每个操作符都会被分配 10 个并行实例。这个设置的优先级高于全局并行度。

算子并行度（Operator-level Parallelism）

你可以在代码中为每个具体的算子设置不同的并行度。Flink 提供了灵活的算子级别并行度控制，可以根据数据处理逻辑的需要对不同的算子设定不同的并行度。例如：

DataStream<String> stream = env.readTextFile("input.txt")                               .map(new MyMapper())                               .setParallelism(5);

在这段代码中，map 操作的并行度被设置为 5，这意味着 map 操作会启动 5 个并发任务来处理数据。其他没有显式设置并行度的算子将使用默认的作业级别并行度。

Slot 并行度（Slot-level Parallelism）

Flink 中的 TaskManager 是执行并行任务的工作节点，每个 TaskManager 中可以包含多个任务槽（Slot）。每个 Slot 对应一个并发任务实例，并可以同时运行多个任务实例。Slot 并行度是 Flink 资源分配中的重要概念，如果作业的并行度超过了集群中可用的 Slot 数，Flink 会进行资源调度，这可能会导致性能下降。每个 TaskManager 可以配置 Slot 数，例如：

taskmanager.numberOfTaskSlots: 4

Operator Level

算子级别，一个算子，数据源和 Sink 并行度可以通过调用 setParalleism（）方法来指定

actions.filter(new FilterFunction<UserAction>() {    @Override    public boolean filter(UserAction value) throws Exception {        return false;    }}).setParallelism(4);

Execution Environment Level

Env 级别

执行环境（任务）的默认并行度可以通过调用 setParallelism()方法指定，为了并行度 3 来执行所有的算子、数据源的 DataSink，可以通过如下的方式设置执行环境的并行度：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(4);

Client Level

客户端级别，推荐使用。并行度可在客户端将 Job 提交到 Flink 时设定，对于 CLI 客户端，可以通过-p 参数指定并行度。

System Level

系统默认级别，尽量不使用。在系统级可以通过设置 flink-conf.yaml 中的 parallism.default 属性来执行环境的默认并行度。

如何设置 Flink 并行度

Flink 提供了几种方法来设置并行度：

在 Flink 配置文件中设置全局并行度

parallelism.default: 4

在提交作业时设置并行度

这里的 -p 20 设置作业的并行度为 20。

flink run -p 20 your-job.jar

在代码中为算子设置并行度

DataStream<String> dataStream = env.readTextFile("input.txt")                                  .map(new MyMapper())                                  .setParallelism(10);

并行度的优化策略

合理设置并行度可以有效提高 Flink 作业的性能，但并行度的设置需要根据数据量、任务复杂度、集群资源等多个因素综合考虑。以下是一些优化策略：

根据数据量设置合理的并行度

对于大数据量的任务，可以通过增加并行度来提高处理速度，但并不是并行度越高越好。过高的并行度会导致资源浪费和任务调度开销。一般来说，建议作业的并行度不要超过 TaskManager 可用 Slot 的总数。

合理分配操作符的并行度

某些操作符，比如 keyBy() 后的 reduce 或 aggregate，其并行度受键值数量的限制，因此为这些操作符设置过高的并行度并不会提高性能。你可以通过数据的特性和操作符的逻辑来合理分配不同操作符的并行度。

利用资源监控进行动态调优

在任务运行时，可以使用 Flink 的 Web UI 来监控作业的运行状态。如果发现某些算子的处理速度慢、资源利用率低，可以考虑调整这些算子的并行度。此外，Flink 允许通过 REST API 或 Web UI 动态调整并行度，而无需重新提交作业。

考虑网络和 I/O 限制

Flink 作业的性能不仅取决于 CPU 和内存，还受限于网络带宽和 I/O 速度。在处理大数据时，如果作业需要频繁地进行网络传输或者 I/O 操作（如读取和写入 HDFS、Kafka），应避免过高的并行度导致网络或磁盘 I/O 的瓶颈。

并行度与容错性

Flink 支持容错机制，当任务失败时，Flink 会根据保存点（checkpoint）进行恢复。高并行度的作业通常会生成更多的 checkpoint 数据，在某些情况下会增加作业恢复时的开销。因此，在设置高并行度时，要同时考虑到 Flink 容错机制可能带来的性能影响。

代码实例

假设有一个作业需要从 Kafka 读取数据，经过 map 转换后将处理结果写入 HDFS。在这种场景下，你可以根据任务的负载和集群资源设置不同算子的并行度：假设我们有一个 Flink 作业，该作业的任务是：

• 从 Kafka 读取实时的交易数据流。

• 对每一条交易数据进行清洗和转换。

• 将清洗后的数据写入 HDFS 进行存储。

这个任务需要根据各个操作的特性设置不同的并行度，以实现性能和资源的最佳利用。

public class FlinkParallelismExample {
    public static void main(String[] args) throws Exception {        // 1. 创建流处理环境        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();                // 设置全局并行度（默认并行度）        env.setParallelism(8);  // 全局默认并行度为8
        // 2. 配置 Kafka 消费者        Properties kafkaProps = new Properties();        kafkaProps.setProperty("bootstrap.servers", "localhost:9092");        kafkaProps.setProperty("group.id", "transaction-group");
        FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<>(                "transaction-topic", new SimpleStringSchema(), kafkaProps);
        // 设置 Kafka 源的并行度        DataStream<String> transactionStream = env                .addSource(kafkaSource)                .setParallelism(6); // 从 Kafka 读取数据时的并行度为 6
        // 3. 数据转换操作        DataStream<String> cleanedData = transactionStream                .map(value -> cleanTransactionData(value))                .setParallelism(12); // 数据清洗的并行度为 12
        // 4. 将清洗后的数据写入 HDFS        cleanedData                .writeAsText("hdfs://namenode:8020/flink/cleaned_transactions/")                .setParallelism(4);  // 写入 HDFS 的并行度为 4
        // 5. 启动任务        env.execute("Flink Parallelism Example");    }
    // 数据清洗的逻辑    public static String cleanTransactionData(String transaction) {        // 假设清洗逻辑包括去除不必要的字段，格式化数据等        return transaction.trim();  // 简单清洗逻辑示例    }}

代码说明 - 并行度配置详解

1. 全局并行度配置

我们在代码中通过 env.setParallelism(8) 设置了全局的并行度为 8。这个配置会作为默认值应用到所有算子，除非某个算子显式设置了不同的并行度。例如：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(8);  // 设置全局默认并行度

这意味着：

• 所有未明确设置并行度的算子都会自动使用 8 个并行任务

• 这个值应该根据集群可用的 CPU 核心数来合理设置

• 在 YARN 集群部署时，通常应该小于等于 YARN 容器分配的 vcore 总数

2. Kafka 消费并行度

通过 setParallelism(6) 为 Kafka 消费者设置了专门的并行度：

KafkaSource<String> source = KafkaSource.<String>builder()    .setTopics("transaction-topic")    .build();DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source")    .setParallelism(6);  // 设置Kafka消费并行度

关键考虑因素：

• Kafka 分区数与并行度的关系：最佳实践是让并行度等于或小于 Kafka 分区数

• 如果 Kafka 有 6 个分区，设置 6 个并行任务可以确保每个分区有一个专用消费者

• 超过分区数的并行度会导致部分任务处于空闲状态

• 消费性能：增加并行度可以提高吞吐量，但受限于：

• 网络带宽

• Kafka broker 处理能力

• 消费者组协调开销

3. 数据转换并行度

在 map 操作阶段，我们设置了更高的并行度(12)：

DataStream<Transaction> processed = stream    .map(new TransactionParser())    .setParallelism(12);  // 设置转换操作并行度

设计考虑：

• 计算密集型操作：如果 map 函数包含复杂计算，增加并行度可以缩短处理时间

• 资源利用：需要确保：

• 集群有足够的 slot/taskmanager 支持这个并行度

• 不会导致内存不足或频繁 GC

• 数据倾斜处理：对于可能产生数据倾斜的转换，更高并行度有助于分散负载

• 与上游并行度的关系：从 6 增加到 12 会产生数据重分发(redistribute)开销

4. HDFS 写入并行度

最终写入阶段设置了较低并行度(4)：

processed.addSink(new HdfsSink())    .setParallelism(4);  // 设置HDFS写入并行度

原因分析：

• I/O 瓶颈：HDFS 写入受磁盘 I/O 限制，过高并行度会导致：

• 磁盘争用

• NameNode 压力增加

• 可能触发 HDFS 的写入流控

• 写入合并：适度的并行度有利于小文件合并

• 检查点影响：写入并行度影响检查点完成时间

• 建议实践：

• 监控 HDFS datanode 的磁盘利用率

• 根据 HDFS 集群规模调整

• 考虑使用批量写入而非单条记录写入

并行度设置最佳实践总结

1. 从数据源并行度开始设计，特别是 Kafka/消息队列等

2. 针对不同类型的操作采用差异化并行度：

3. I/O 密集型：适中或较低并行度

4. CPU 密集型：较高并行度

5. 考虑算子链(operator chain)对并行度的影响

6. 监控反压(backpressure)指标调整并行度

7. 在测试环境进行并行度基准测试