Flink 1.17教程：运行时架构详解

桃扇骨 2023-10-15 14:39 56阅读 0赞

## 运行时架构——Standalone会话模式为例 ##

![img][]

## 并行度 ##

并行度是指在计算过程中同时执行多个任务或操作的能力。在 Apache Flink 中，并行度是指同时执行作业中的多个任务或算子的能力。

并行度的引入是为了解决以下问题：

1.  提高计算速度：通过将任务拆分成多个子任务，并行执行它们，可以大大提高计算速度。每个子任务都可以在不同的计算资源上独立执行，从而充分利用计算资源的并行处理能力。
2.  处理大规模数据：当需要处理大规模数据时，单个任务可能会面临内存不足或处理时间过长的问题。通过将任务分成多个并行任务，可以将负载均衡到多个计算资源上，提高处理大规模数据的效率和可扩展性。
3.  提高系统的容错性：并行度还可以提高系统的容错性。如果一个任务或算子失败，其他并行任务仍然可以继续执行，从而减少了整体作业的影响。

总的来说，通过并行度，可以将任务分解成更小的单元并同时执行它们，提高计算速度、处理大规模数据和提高系统的容错性。这是为了更好地利用计算资源、提高系统性能和处理大规模数据而引入的概念。

![img][img 1]

一个特定算子的子任务（subtask）的个数被称之为其并行度（parallelism）。这样，包含并行子任务的数据流，就是并行数据流，它需要多个分区（stream partition）来分配并行任务。一般情况下，一个流程序的并行度，可以认为就是其所有算子中最大的并行度。一个程序中，不同的算子可能具有不同的并行度。  
例如：如上图所示，当前数据流中有source、map、window、sink四个算子，其中sink算子的并行度为1，其他算子的并行度都为2。所以这段流处理程序的并行度就是2。

## 并行度设置&优先级 ##

### 并行度（Parallelism） ###

#### 并行度的设置 ####

在Flink中，可以用不同的方法来设置并行度，它们的有效范围和优先级别也是不同的。

#### 代码中设置 ####

我们在代码中，可以很简单地在算子后跟着调用setParallelism()方法，来设置当前算子的并行度：

stream.map(word -> Tuple2.of(word, 1L)).setParallelism(2);

这种方式设置的并行度，只针对当前算子有效。

另外，我们也可以直接调用执行环境的setParallelism()方法，全局设定并行度：

env.setParallelism(2);

这样代码中所有算子，默认的并行度就都为2了。我们一般不会在程序中设置全局并行度，因为如果在程序中对全局并行度进行硬编码，会导致无法动态扩容。

这里要注意的是，由于`keyBy`不是算子，所以无法对keyBy设置并行度。

#### 提交应用时设置 ####

在使用flink run命令提交应用时，可以增加-p参数来指定当前应用程序执行的并行度，它的作用类似于执行环境的全局设置：

bin/flink run –p 2 –c com.atguigu.wc.SocketStreamWordCount ./FlinkTutorial-1.0-SNAPSHOT.jar

如果我们直接在Web UI上提交作业，也可以在对应输入框中直接添加并行度。  
![img][img 2]

![img][img 3]

![img][img 4]

![img][img 5]![img][img 6]

![img][img 7]

package com.atguigu.wc;
     
    import org.apache.flink.api.common.typeinfo.TypeHint;
    import org.apache.flink.api.common.typeinfo.TypeInformation;
    import org.apache.flink.api.common.typeinfo.Types;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.util.Collector;
     
    /**
     * TODO DataStream实现Wordcount：读socket（无界流）
     *
     * @author
     * @version 1.0
     */
    public class WordCountStreamUnboundedDemo {
        
        public static void main(String[] args) throws Exception {
        
            // TODO 1.创建执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            // IDEA运行时，也可以看到webui，一般用于本地测试
            // 需要引入一个依赖 flink-runtime-web
            // 在idea运行，不指定并行度，默认就是 电脑的 线程数
            // StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
            env.setParallelism(3);
     
            // TODO 2.读取数据： socket
            DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 7777);
     
            // TODO 3.处理数据: 切换、转换、分组、聚合
            SingleOutputStreamOperator<Tuple2<String, Integer>> sum = socketDS
                    .flatMap(
                            (String value, Collector<Tuple2<String, Integer>> out) -> {
        
                                String[] words = value.split(" ");
                                for (String word : words) {
        
                                    out.collect(Tuple2.of(word, 1));
                                }
                            }
                    )
                    .setParallelism(2)
                    .returns(Types.TUPLE(Types.STRING,Types.INT))
                    // .returns(new TypeHint<Tuple2<String, Integer>>() {})
                    .keyBy(value -> value.f0)
                    .sum(1);
     
            // TODO 4.输出
            sum.print();
     
            // TODO 5.执行
            env.execute();
        }
    }
     
    /**
     并行度的优先级：
        代码：算子 > 代码：env > 提交时指定 > 配置文件
     */

#### 并行度的优先级 ####

代码：算子 > 代码：env > 提交时指定 > 配置文件

## 算子链Operator Chain ##

在 Apache Flink 中，算子链（Operator Chaining）是将多个操作符（算子）连接在一起形成一个链式结构的优化技术。算子链的作用是将多个操作符合并为一个单一的任务单元，以减少通信开销、提高执行效率和减少资源占用。

通俗来说，算子链的作用可以比喻为将多个操作合并成一个整体，就像是把多个小任务捆绑在一起，让它们作为一个整体来执行，而不是分散地执行。

应用场景：

1.  减少通信开销：在一个算子链中，多个操作符可以在同一个线程中执行，避免了数据的序列化和网络通信开销。这对于那些需要频繁传输数据的操作符来说特别有用，可以大大减少数据在不同操作符之间的传输成本。
2.  提高执行效率：算子链可以减少任务切换的开销。在一个算子链中，多个操作符可以在同一个线程中执行，避免了线程之间的切换，并且可以通过内存数据传递来提高执行效率。
3.  减少资源占用：算子链可以减少并行任务的数量，从而减少了需要的计算资源。如果多个操作符在一个算子链中执行，它们可以共享同一个线程或任务资源，这样可以有效地降低整体资源的占用。

总的来说，算子链可以在 Apache Flink 中提供优化的执行方式，减少通信开销、提高执行效率和减少资源占用。它适用于需要频繁传输数据、有依赖关系的操作符，以及需要提高整体执行效率和减少资源消耗的场景。

### 合并算子链 ###

在Flink中，并行度相同的一对一（one to one）算子操作，可以直接链接在一起形成一个“大”的任务（task），这样原来的算子就成为了真正任务里的一部分，如下图所示。每个task会被一个线程执行。这样的技术被称为“算子链”（Operator Chain）。

![img][img 8]

## 算子链演示 ##

1、算子之间的传输关系：  
一对一  
重分区  
2、算子 串在一起的条件：  
1） 一对一  
2） 并行度相同  
3、关于算子链的api：  
1）全局禁用算子链：`env.disableOperatorChaining();`  
2）某个算子不参与链化： `算子A.disableChaining()`, 算子A不会与 前面 和 后面的算子 串在一起  
3）从某个算子开启新链条： `算子A.startNewChain()`， 算子A不与 前面串在一起，从A开始正常链化

package com.atguigu.wc;
     
    import org.apache.flink.api.common.typeinfo.Types;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.util.Collector;
     
    /**
     * TODO DataStream实现Wordcount：读socket（无界流）
     *
     * @author
     * @version 1.0
     */
    public class OperatorChainDemo {
        
        public static void main(String[] args) throws Exception {
        
            // TODO 1.创建执行环境
    		// StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            // IDEA运行时，也可以看到webui，一般用于本地测试
            // 需要引入一个依赖 flink-runtime-web
            StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
     
            // 在idea运行，不指定并行度，默认就是 电脑的 线程数
            env.setParallelism(1);
     
            // 全局禁用 算子链
    		//env.disableOperatorChaining();
     
            // TODO 2.读取数据：socket
            DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 7777);
     
            // TODO 3.处理数据: 切换、转换、分组、聚合
            SingleOutputStreamOperator<Tuple2<String,Integer>> sum = socketDS
    				//.disableChaining()
                    .flatMap(
                            (String value, Collector<String> out) -> {
        
                                String[] words = value.split(" ");
                                for (String word : words) {
        
                                    out.collect(word);
                                }
                            }
                    )
                    .startNewChain()
    				//.disableChaining()
                    .returns(Types.STRING)
                    .map(word -> Tuple2.of(word, 1))
                    .returns(Types.TUPLE(Types.STRING,Types.INT))
                    .keyBy(value -> value.f0)
                    .sum(1);
     
            // TODO 4.输出
            sum.print();
     
            // TODO 5.执行
            env.execute();
        }
    }
     
    /**
     1、算子之间的传输关系：
         一对一
         重分区
     2、算子 串在一起的条件：
        1） 一对一
        2） 并行度相同
     3、关于算子链的api：
        1）全局禁用算子链：env.disableOperatorChaining();
        2）某个算子不参与链化：  算子A.disableChaining(),  算子A不会与 前面 和 后面的算子 串在一起
        3）从某个算子开启新链条：  算子A.startNewChain()， 算子A不与 前面串在一起，从A开始正常链化
     */

## 任务槽Task Slots ##

在 Apache Flink 中，任务槽（Task Slots）是指可用于执行并行任务的资源单元。每个任务槽可以看作是一个可用的执行线程或处理单元，用于并行执行作业的不同部分。

通俗来说，可以将任务槽想象成一个工作台，而每个工作台上都可以同时进行一项任务。任务槽的数量决定了同时可以执行的任务数量。

任务槽的作用和应用场景：

1.  并行执行：任务槽允许在同一时刻并行执行多个任务。每个任务槽可以独立地执行一个任务或算子，从而有效地利用计算资源，提高作业的并行度和整体处理能力。
2.  任务分配和负载均衡：任务槽可以用于将不同的任务或算子分配到不同的资源上。通过合理分配任务槽，可以实现负载均衡，确保任务在可用资源之间均匀分配，避免资源浪费和瓶颈。
3.  容错和高可用性：任务槽还可以提供容错和高可用性。如果一个任务槽上的任务或算子失败，系统可以将其重新分配到其他可用的任务槽上，从而实现故障恢复和保证作业的持续执行。

总的来说，任务槽在 Apache Flink 中扮演着并行执行任务的角色，可以提高作业的并行度和整体处理能力。它们用于任务分配、负载均衡以及容错和高可用性的实现。

![img][img 9]

## 任务槽的共享组 ##

![img][img 10]

默认情况下，Flink是允许子任务共享slot的。如果我们保持sink任务并行度为1不变，而作业提交时设置全局并行度为6，那么前两个任务节点就会各自有6个并行子任务，整个流处理程序则有13个子任务。如上图所示，只要属于同一个作业，那么对于不同任务节点（算子）的并行子任务，就可以放到同一个slot上执行。所以对于第一个任务节点source→map，它的6个并行子任务必须分到不同的slot上，而第二个任务节点keyBy/window/apply的并行子任务却可以和第一个任务节点共享slot。

package com.atguigu.wc;
     
    import org.apache.flink.api.common.typeinfo.Types;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.util.Collector;
     
    /**
     * TODO DataStream实现Wordcount：读socket（无界流）
     *
     * @author
     * @version 1.0
     */
    public class SlotSharingGroupDemo {
        
        public static void main(String[] args) throws Exception {
        
            // TODO 1.创建执行环境
    		// StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            // IDEA运行时，也可以看到webui，一般用于本地测试
            // 需要引入一个依赖 flink-runtime-web
            StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
     
            // 在idea运行，不指定并行度，默认就是 电脑的 线程数
            env.setParallelism(1);
     
            // TODO 2.读取数据：socket
            DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 7777);
     
            // TODO 3.处理数据: 切换、转换、分组、聚合
            SingleOutputStreamOperator<Tuple2<String,Integer>> sum = socketDS
                    .flatMap(
                            (String value, Collector<String> out) -> {
        
                                String[] words = value.split(" ");
                                for (String word : words) {
        
                                    out.collect(word);
                                }
                            }
                    )
                    .returns(Types.STRING)
                    .map(word -> Tuple2.of(word, 1)).slotSharingGroup("aaa")
                    .returns(Types.TUPLE(Types.STRING,Types.INT))
                    .keyBy(value -> value.f0)
                    .sum(1);
     
     
            // TODO 4.输出
            sum.print();
     
            // TODO 5.执行
            env.execute();
        }
    }
     
    /**
     1、slot特点：
        1）均分隔离内存，不隔离cpu
        2）可以共享：
              同一个job中，不同算子的子任务 才可以共享 同一个slot，同时在运行的前提是，属于同一个 slot共享组，默认都是“default”
     2、slot数量 与 并行度 的关系
        1）slot是一种静态的概念，表示最大的并发上限
           并行度是一种动态的概念，表示 实际运行 占用了 几个
        2）要求： slot数量 >= job并行度（算子最大并行度），job才能运行
           TODO 注意：如果是yarn模式，动态申请
             --> TODO 申请的TM数量 = job并行度 / 每个TM的slot数，向上取整
           比如session： 一开始 0个TaskManager，0个slot
             --> 提交一个job，并行度10
                --> 10/3，向上取整，申请4个tm，
                --> 使用10个slot，剩余2个slot
     */

## slot与并行度的关系&演示 ##

### 任务槽和并行度的关系 ###

任务槽和并行度都跟程序的并行执行有关，但两者是完全不同的概念。简单来说任务槽是静态的概念，是指`TaskManager`具有的并发执行能力，可以通过参数`taskmanager.numberOfTaskSlots`进行配置；而并行度是动态概念，也就是`TaskManager`运行程序时实际使用的并发能力，可以通过参数`parallelism.default`进行配置。

1、slot特点：  
1）均分隔离内存，不隔离cpu  
2）可以共享：  
同一个job中，不同算子的子任务 才可以共享 同一个slot，同时在运行的前提是，属于同一个 slot共享组，默认都是“default”  
2、slot数量 与 并行度 的关系  
1）slot是一种静态的概念，表示最大的并发上限  
并行度是一种动态的概念，表示 实际运行 占用了 几个  
2）要求： slot数量 >= job并行度（算子最大并行度），job才能运行  
注意：如果是yarn模式，动态申请  
\--> 申请的TM数量 = job并行度 / 每个TM的slot数，向上取整  
比如session： 一开始 0个TaskManager，0个slot  
\--> 提交一个job，并行度10  
\--> 10/3，向上取整，申请4个tm，  
\--> 使用10个slot，剩余2个slot

并行度是指在作业执行过程中同时执行的任务或算子的数量。它决定了作业可以同时处理多少个数据分片或并行操作。并行度可以在作业级别、算子级别或子任务级别进行设置。

任务槽则是实际的执行资源单元，用于并行执行任务或算子。每个任务槽可以看作一个可用的执行线程或处理单元。

关系如下：

1.  并行度和任务槽数量：并行度通常与任务槽的数量相关联。如果并行度大于任务槽的数量，那么任务将被分配到多个任务槽中执行，从而实现并行处理。如果并行度小于或等于任务槽的数量，那么每个任务或算子可以在一个单独的任务槽中执行。
2.  负载均衡：任务槽的数量可以用于实现任务的负载均衡。如果并行度大于任务槽的数量，系统会尽量将任务均匀分配到可用的任务槽中，以实现资源的充分利用和负载的平衡。
3.  容错性：任务槽的数量也与容错性相关。如果某个任务槽上的任务或算子失败，系统可以将其重新分配到其他可用的任务槽上，从而实现故障恢复和保证作业的持续执行。

综上所述，任务槽和并行度密切相关，任务槽提供了实际的执行资源单元，用于并行执行任务或算子。并行度决定了同时执行的任务或算子的数量，而任务槽的数量可以影响并行度的设置、负载均衡和容错性。

## 提交流程\_Standalone会话模式 ##

### Standalone会话模式作业提交流程 ###

![img][img 11]

### 逻辑流图/作业图/执行图/物理流图 ###

逻辑流图（StreamGraph）→ 作业图（JobGraph）→ 执行图（ExecutionGraph）→ 物理图（Physical Graph）。

![img][img 12]

![img][img 13]

## 提交流程\_Yarn应用模式 ##

Yarn应用模式作业提交流程：

![img][img 14]

--------------------

links：

[https://upward.blog.csdn.net/article/details/131760766][https_upward.blog.csdn.net_article_details_131760766]

--------------------

links：

[https://upward.blog.csdn.net/article/details/131760766][https_upward.blog.csdn.net_article_details_131760766]

[img]: https://img-blog.csdnimg.cn/73f9e3ebb5a0416d8fb741de31dd5c8c.png
[img 1]: https://img-blog.csdnimg.cn/6295e55cf2bc444f873ad2d11c419dd7.png
[img 2]: https://img-blog.csdnimg.cn/34cd5232a1b0443f9f10361def994c05.png
[img 3]: https://img-blog.csdnimg.cn/8147c5f6deea4b3780291f85934ef521.png
[img 4]: https://img-blog.csdnimg.cn/dd6e694d5b234905a091954131cf8f12.png
[img 5]: https://img-blog.csdnimg.cn/20656e5fc29244e79f3eaa3a2be8c8a8.png
[img 6]: https://img-blog.csdnimg.cn/0ecead20696741f798b14bbb21db6d4d.png
[img 7]: https://img-blog.csdnimg.cn/4983b0a0b6824f81a0ec9eccc6f54474.png
[img 8]: https://img-blog.csdnimg.cn/d7248b5fa87140d58b02488a33cd40f1.png
[img 9]: https://img-blog.csdnimg.cn/5533aafc346a4ebe8c1b1ad14e44573d.png
[img 10]: https://img-blog.csdnimg.cn/82f423d2949a4a37990205284e3cf32f.png
[img 11]: https://img-blog.csdnimg.cn/dbc2845e7de544a6a8a715dd27da882f.png
[img 12]: https://img-blog.csdnimg.cn/7a7b34b8a09f48549af42ab4a103734a.png
[img 13]: https://img-blog.csdnimg.cn/2b16a038b15d48e7a3647c813b977aff.png
[img 14]: https://img-blog.csdnimg.cn/3413a7d6dfca432291e20163e705f0ac.png
[https_upward.blog.csdn.net_article_details_131760766]: https://upward.blog.csdn.net/article/details/131760766