点一下关注吧！！！非常感谢！！持续更新！！！
目前已经更新到了：
- 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 集群的默认全局并行度：
这个配置将为每个没有指定并行度的操纵符分配默认的 4 个并行实例。假如你没有在代码中或任务提交时明白设置并行度，Flink 将使用这个默认值。
作业并行度（Job-level Parallelism）
在提交 Flink 作业时，你可以为整个作业设置并行度，覆盖全局默认值。例如，在命令利用用 flink run 提交作业时可以通过 -p 参数来设置并行度：
- flink run -p 10 your-job.jar
算子并行度（Operator-level Parallelism）
你可以在代码中为每个具体的算子设置差别的并行度。Flink 提供了灵活的算子级别并行度控制，可以根据数据处理逻辑的需要对差别的算子设定差别的并行度。例如：
- DataStream<String> stream = env.readTextFile("input.txt")
- .map(new MyMapper())
- .setParallelism(5);
Slot 并行度（Slot-level Parallelism）
Flink 中的 TaskManager 是执行并行任务的工作节点，每个 TaskManager 中可以包含多个任务槽（Slot）。每个 Slot 对应一个并发任务实例，并可以同时运行多个任务实例。Slot 并行度是 Flink 资源分配中的重要概念，假如作业的并行度凌驾了集群中可用的 Slot 数，Flink 会举行资源调度，这可能会导致性能降落。
每个 TaskManager 可以配置 Slot 数，例如：
- taskmanager.numberOfTaskSlots: 4
算子级别，一个算子，数据源和Sink并行度可以通过调用setParalleism（）方法来指定
- actions.filter(new FilterFunction<UserAction>() {
- @Override
- public boolean filter(UserAction value) throws Exception {
- return false;
- }
- }).setParallelism(4);
Env级别
执行环境（任务）的默认并行度可以通过调用setParallelism()方法指定，为了并行度3来执行所有的算子、数据源的DataSink，可以通过如下的方式设置执行环境的并行度：
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setParallelism(4);
客户端级别，推荐使用。
并行度可在客户端将Job提交到Flink时设定，对于CLI客户端，可以通过-p参数指定并行度。
System Level
体系默认级别，只管不使用。在体系级可以通过设置 flink-conf.yaml中的parallism.default属性来执行环境的默认并行度。
如何设置 Flink 并行度
Flink 提供了几种方法来设置并行度：
在 Flink 配置文件中设置全局并行度
在提交作业时设置并行度
这里的 -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是动态的概念，是指定程序运行时实际使用的并发能力
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。 |
