大数据-128 - Flink 并行度设置 细节详解 全局、作业、算子、Slot

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

目前已经更新到了：

• Hadoop（已更完）
• HDFS（已更完）
• MapReduce（已更完）
• Hive（已更完）
• Flume（已更完）
• Sqoop（已更完）
• Zookeeper（已更完）
• HBase（已更完）
• Redis （已更完）
• Kafka（已更完）
• Spark（已更完）
• Flink（正在更新！）

章节内容

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

• ManageOperatorState
• StateBackend
• Checkpoint

简单介绍

一个Flink程序由多个Operator组成（Source、Transformation、Sink）。
一个Operator由多个并行的Task（线程）来执行，一个Operator的并行Task（线程）数目就被称为该Operator（任务）并行度（Paralle）
并行度可以有如下几种指定方式。
Flink 中的并行度（Parallelism）是指每个算子（Operator）在任务执行时可以同时处理数据的并发实例数。Flink 的核心优势之一就是能够通过并行处理大规模数据来提高效率和性能。通过正确设置并行度，你可以在充分利用集群资源的同时，实现高效的数据处理。接下来详细介绍 Flink 并行度的概念、设置方法及其优化策略。

并行度的概念

在 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 进行存储。
  这个任务需要根据各个操作的特性设置不同的并行度，以实现性能和资源的最佳利用。

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.streaming.connectors.kafka.KafkaSink;
import org.apache.flink.core.fs.Path;

import java.util.Properties;

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。也就是说，Flink 将会启动 6 个并行任务来从Kafka 的 transaction-topic 主题中消费数据。这个并行度可以根据 Kafka 分区的数量调整。如果 Kafka 有 6 个分区，那么设置并行度为 6 是合理的，这样可以保证每个分区都有一个并发实例进行处理。
• 数据转换并行度：数据从 Kafka 读取后，进入 map 操作进行清洗和转换。这里的并行度被设置为 12（setParallelism(12)），即清洗任务将启动 12 个并行实例来同时处理数据。这可以提高数据处理速度，但也需要确保集群中有足够的计算资源支持这个并行度。
• HDFS 写入并行度：在数据清洗完成后，将数据写入 HDFS 文件系统。这里我们设置了写入 HDFS 的并行度为 4（setParallelism(4)）。这意味着将有 4 个并发任务负责将数据写入到 HDFS。由于 HDFS 的写入通常涉及磁盘 I/O 操作，设置较低的并行度可以避免 I/O 争用。

注意

• 并行度优先级：算计级别 > env级别 > Client 级别 > 系统默认级别
• 如果Source不可以被并行执行，即使指定了并行度为多个，也不会生效
• 尽可能的规避算子的并行度的设置，因为并行度的改变会造成Task的重新划分，带来Shuffle问题
• 推荐使用任务提交的时候动态的指定并行度
• slot是静态的概念，是指TaskManager具有的并发执行能力：parallelism是动态的概念，是指定程序运行时实际使用的并发能力