Flink 并行度的设置

在 Apache Flink 中，并行度（Parallelism） 是控制任务并发执行的核心参数之一。Flink 提供了 多个层级设置并行度的方式，优先级从高到低如下：

🧩 一、Flink 并行度的四个设置层级

✅ 二、各层级设置详解与示例

1. Operator Level（算子级别）

• 优先级最高
• 可以为特定算子设置不同并行度，适用于数据倾斜或资源敏感操作

🔧 示例：

DataStream<String> stream = env.fromElements("a", "b", "c");

// 单独为 map 算子设置并行度为4
stream.map(new MyMapFunction())
      .setParallelism(4)
      .print();

✅ 适用场景：

• 某个算子计算密集，需要更多资源
• 数据源分区数较少，但后续算子可并行化处理

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

• 设置整个 Job 的默认并行度
• 如果未对某些算子单独设置，并使用此值

🔧 示例：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4); // 所有算子默认并行度为4

DataStream<String> stream = env.fromElements("a", "b", "c");
stream.map(new MyMapFunction()).print(); // 默认并行度为4

✅ 适用场景：

• 多数算子使用相同并行度
• 统一配置便于管理和维护

3. Client Level（客户端提交作业时）

• 使用命令行参数动态设置并行度
• 不修改代码即可适配不同运行环境（如测试/生产）

🔧 示例：

flink run -p 4 -c com.example.MyJob ./myjob.jar

✅ 适用场景：

• 快速调整不同集群资源配置
• 测试阶段快速验证性能

4. System Level（系统级别）

• 在 flink-conf.yaml 中设置全局默认并行度
• 对所有提交的作业生效（除非被更高级别覆盖）

🔧 示例（flink-conf.yaml）：

parallelism.default: 4

✅ 适用场景：

• 所有作业共享相同的默认资源配置
• 避免手动重复设置

📊 三、并行度优先级对比表

💡 四、并行度设置建议

✅ 推荐做法：

• 开发/测试环境：使用 .setParallelism() 或 -p 命令行设置较小值（如1~4）
• 生产环境：
  • 使用 flink-conf.yaml 设置基础并行度
  • 使用 env.setParallelism() 明确控制默认值
  • 为关键算子单独设置更高并行度（如窗口聚合、复杂逻辑）

⚙️ 示例组合：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4); // 默认并行度

env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source")
   .setParallelism(8) // Kafka Source 并行度设为8（等于topic分区数）
   .map(new MyMapFunction()) // 使用默认并行度4
   .keyBy(keySelector)
   .window(TumblingEventTimeWindows.of(Time.seconds(5)))
   .process(new MyProcessWindowFunction()) // 可选 setParallelism()
   .print();

🧠 五、并行度与资源的关系

✅ 建议：确保总并行度 ≤ 总 slot 数量

📈 六、实际调优建议

✅ 七、完整示例（Java + Shell）

Java 设置（Env + Operator）：

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

env.fromElements("a", "b", "c")
   .map(x -> x)
   .setParallelism(2) // 覆盖默认值
   .print();

env.execute("Parallelism Example");

Shell 设置（Client Level）：

flink run -p 8 -c com.example.MyJob ./myjob.jar

✅ 八、总结