并行度的基本概念

在Apache Flink中，并行度（Parallelism）是指每个算子（Operator）在执行时可以同时运行的并行任务（Task）数量。Flink程序的数据流图由多个算子组成（包括Source、Transformation和Sink），每个算子都可以独立设置其并行度。

并行度的重要性

并行度是Flink性能调优的关键参数之一：

  • 决定了任务在集群中的资源使用情况
  • 直接影响数据处理吞吐量
  • 影响作业的容错能力和恢复速度

并行度的组成结构

Flink程序的执行模型可以分解为：

  1. Job: 完整的Flink程序
  2. Operator: 数据处理的基本单元（如map、filter、join等）
  3. Task: 算子并行执行的实例
  4. Subtask: 实际的执行单元（线程）

并行度的设置方式

1. 全局并行度设置

// 在 ExecutionConfig 中设置
env.setParallelism(4);
  • 影响整个作业中所有算子的默认并行度
  • 通常在作业入口处设置

2. 算子级别并行度

DataStream<String> data = env.addSource(new CustomSource())
    .map(new MyMapper()).setParallelism(2)
    .filter(new MyFilter()).setParallelism(3);
  • 可以针对特定算子设置与全局不同的并行度
  • 适用于计算密集型或I/O密集型的特殊算子

3. 配置文件设置

在flink-conf.yaml中：

parallelism.default: 3
  • 作为集群的默认并行度
  • 当代码中未明确设置时使用此值

4. 客户端提交参数

./bin/flink run -p 5 -c com.example.MyJob myJob.jar
  • 在使用CLI提交作业时指定
  • 会覆盖代码中的全局并行度设置

并行度设置的最佳实践

考虑因素

  1. 数据量大小：大数据量通常需要更高并行度
  2. 算子特性：像window这样的有状态算子可能需要特殊考虑
  3. 集群资源：不超过可用TaskManager的slot总数
  4. 数据倾斜：可能需要调整并行度来平衡负载

典型场景示例

  • ETL管道：Source和Sink并行度通常与外部系统分区数匹配
  • 流式分析：关键算子可能需要2-4倍CPU核心数的并行度
  • 批处理作业：可设置较高并行度以加快处理速度

并行度的层级

全局并行度（Global Parallelism）

全局并行度是指Flink集群默认为所有作业和操作符分配的并行度。在配置文件flink-conf.yaml中：

parallelism.default: 4

作业并行度（Job-level Parallelism）

在提交Flink作业时，通过-p参数设置：

flink run -p 10 your-job.jar

算子并行度（Operator-level Parallelism）

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

Slot 并行度（Slot-level Parallelism）

每个TaskManager可以配置Slot数：

taskmanager.numberOfTaskSlots: 4

并行度的优化策略

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

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

合理分配操作符的并行度

某些操作符，比如keyBy()后的reduce或aggregate，其并行度受键值数量的限制，因此为这些操作符设置过高的并行度并不会提高性能。

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

在任务运行时，可以使用Flink的Web UI来监控作业的运行状态。如果发现某些算子的处理速度慢、资源利用率低，可以考虑调整这些算子的并行度。

考虑网络和 I/O 限制

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

并行度与容错性

Flink支持容错机制，当任务失败时，Flink会根据保存点（checkpoint）进行恢复。高并行度的作业通常会生成更多的checkpoint数据，在某些情况下会增加作业恢复时的开销。

代码实例

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();  // 简单清洗逻辑示例
    }
}

代码说明

  • 全局并行度：通过env.setParallelism(8)设置了全局的并行度为8。这意味着，除非显式设置，所有的算子默认都会使用8个并行实例运行。

  • Kafka消费并行度：通过setParallelism(6)为从Kafka读取数据的操作设置了并行度为6。这个并行度可以根据Kafka分区的数量调整。如果Kafka有6个分区，那么设置并行度为6是合理的。

  • 数据转换并行度：并行度被设置为12（setParallelism(12)），即清洗任务将启动12个并行实例来同时处理数据。这可以提高数据处理速度，但也需要确保集群中有足够的计算资源支持这个并行度。

  • HDFS写入并行度：设置了写入HDFS的并行度为4（setParallelism(4)）。由于HDFS的写入通常涉及磁盘I/O操作，设置较低的并行度可以避免I/O争用。

并行度优先级规则

Flink任务的并行度设置遵循明确的优先级顺序：

  1. 算子级别 - 直接在代码中为特定算子设置（最高优先级）
  2. 环境级别 - 通过env.setParallelism()设置
  3. 客户端级别 - 提交任务时通过参数指定
  4. 系统默认级别 - 使用flink-conf.yaml中的默认配置（最低优先级）

并行度限制条件

  • Source并行度限制：如果数据源（如Kafka特定分区或单文件）不支持并行读取，即使设置了高并行度也不会生效
  • 示例：读取单个CSV文件时，并行度只能为1

并行度调整注意事项

  1. Shuffle影响：改变并行度会导致任务重新划分，可能引发数据Shuffle

    • 频繁调整会增加网络开销
    • 可能导致数据倾斜问题

  2. 推荐做法：

    • 优先使用任务提交时动态指定（如-p参数）
    • 在生产环境中，建议通过-p参数统一控制

核心概念区分

  • Slot（槽位）：

    • 静态资源概念
    • 代表TaskManager的实际并发执行能力
    • 配置方式：taskmanager.numberOfTaskSlots

  • Parallelism（并行度）：

    • 动态执行概念
    • 表示程序运行时实际使用的并发数量
    • 可通过代码或参数动态调整

最佳实践建议

  1. 对于批处理作业，可在不同阶段设置不同并行度
  2. 流处理作业建议保持稳定的并行度设置
  3. 监控并行度调整后的资源使用情况
  4. 考虑下游系统的处理能力（如数据库连接池大小）