Flink实战（十）Flink on Yarn模式

「爱情、让人受尽委屈。」 2023-07-21 06:14 108阅读 0赞

### 文章目录 ###

*   *  概述
     *   *  模式1 Yarn Per Job
         *   *  实践
         *  模式2 Yarn On Session
         *   *  实践
         *  参考资料

## 概述 ##

![img][]

FLink 如何执行在Yarn上面的呢？

首先我们需要了解到Flink的底层执行流程，然后看哪一步与Yarn进行适配，执行。

**Flink底层执行流程：**

1.  用户通过 DataStream API、DataSet API、SQL 和 Table API 编写 Flink 任务，它会生成一个JobGraph。
2.  JobGraph 是由 source、map()、keyBy()/window()/apply() 和 Sink 等算子组成的。
3.  当 JobGraph 提交给 Flink 集群后，能够以 Local、Standalone、Yarn 和 Kubernetes 四种模式运行。

*那么我们下面主要看一下运行在Yarn的两种模式：*

### 模式1 Yarn Per Job ###

![img][img 1]

Flink on Yarn 中的 Per Job 模式是指每次提交一个任务，然后**任务运行完成**之后资源就会被**释放**。

*  首先 Client 提交 Yarn App，比如 JobGraph 或者 JARs。
 *  接下来 Yarn 的 ResourceManager 会申请第一个 Container。这个 Container 通过 Application Master 启动进程，Application Master 里面运行的是 Flink 程序，即 Flink-Yarn ResourceManager 和 JobManager。
 *  最后 Flink-Yarn ResourceManager 向 Yarn ResourceManager 申请资源。当分配到资源后，启动 TaskManager。TaskManager 启动后向 Flink-Yarn ResourceManager 进行注册，注册成功后 JobManager 就会分配具体的任务给 TaskManager 开始执行。

#### 实践 ####

# 默认：{masterMemoryMB=1024, taskManagerMemoryMB=1024,numberTaskManagers=1, slotsPerTaskManager=1}
    
    ./bin/yarn-session.sh
    
    ## 启动一个有2个Taskmanager，jobmanager内存1GB，taskManager1GB内存的集群
    ./bin/yarn-session.sh -n 2 -jm 1024 -tm 1024
    
    ## 后台模式
    ./bin/yarn-session.sh -n 2 -jm 1024 -tm 1024 -d

然后在看输出日志里面

Flink JobManager is now running on flinkhadoop:43434 with leader id 00000000-0000-0000-0000-000000000000.
    JobManager Web Interface: http://flinkhadoop:43434

本地调试的时候，直接把对应的执行信息改变即可！

final StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("flinkhadoop", 43434,
                    "jar包位置");

**参考：**

Usage:
       Required
         -n,--container <arg>   Number of YARN container to allocate (=Number of Task Managers)
       Optional
         -at,--applicationType <arg>     Set a custom application type for the application on YARN
         -D <property=value>             use value for given property
         -d,--detached                   If present, runs the job in detached mode
         -h,--help                       Help for the Yarn session CLI.
         -id,--applicationId <arg>       Attach to running YARN session
         -j,--jar <arg>                  Path to Flink jar file
         -jm,--jobManagerMemory <arg>    Memory for JobManager Container with optional unit (default: MB)
         -m,--jobmanager <arg>           Address of the JobManager (master) to which to connect. Use this flag to connect to a different JobManager than the one specified in the configuration.
         -n,--container <arg>            Number of YARN container to allocate (=Number of Task Managers)
         -nl,--nodeLabel <arg>           Specify YARN node label for the YARN application
         -nm,--name <arg>                Set a custom name for the application on YARN
         -q,--query                      Display available YARN resources (memory, cores)
         -qu,--queue <arg>               Specify YARN queue.
         -s,--slots <arg>                Number of slots per TaskManager
         -sae,--shutdownOnAttachedExit   If the job is submitted in attached mode, perform a best-effort cluster shutdown when the CLI is terminated abruptly, e.g., in response to a user interrupt, such
                                         as typing Ctrl + C.
         -st,--streaming                 Start Flink in streaming mode
         -t,--ship <arg>                 Ship files in the specified directory (t for transfer)
         -tm,--taskManagerMemory <arg>   Memory per TaskManager Container with optional unit (default: MB)
         -yd,--yarndetached              If present, runs the job in detached mode (deprecated; use non-YARN specific option instead)
         -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths for high availability mode

系统默认使用`con/flink-conf.yaml`里的配置。Flink on yarn将会覆盖掉几个参数：

`jobmanager.rpc.address`因为jobmanager的在集群的运行位置并不是实现确定的，前面也说到了就是am的地址；

`taskmanager.tmp.dirs`使用yarn给定的临时目录;

`parallelism.default`也会被覆盖掉，如果在命令行里指定了slot数。

如果你想保证conf/flink-conf.yaml仅是全局默认配置，然后针对要启动的每一个yarn-session.sh都设置自己的配置，那么可以考虑使用-D修饰。

### 模式2 Yarn On Session ###

![img][img 2]

在 Per Job 模式中，执行完任务后整个资源就会释放，包括 JobManager、TaskManager 都**全部退出**。而 Session 模式则不一样，它的 **Dispatcher 和 ResourceManager** 是可以复用的。

**Session 模式**下，当 Dispatcher 在收到请求之后，会启动 `JobManager(A)`，让 `JobManager(A)` 来完成启动 TaskManager，接着会启动`JobManager(B)`和对应的 `TaskManager` 的运行。当 A、B 任务运行完成后，资源并不会释放。`Session 模式`也称为**多线程模式**，其特点是资源会一直存在不会释放，多个 JobManager **共享**一个 `Dispatcher`，而且还共享 `Flink-YARN ResourceManager`。

Session 模式和 Per Job 模式的应用场景不一样。Per Job 模式比较适合那种**对启动时间不敏感**，**运行时间较长**的任务。Seesion 模式适合**短时间运行的任务**，一般是**批处理任务**。若用 Per Job 模式去运行短时间的任务，那就需要频繁的申请资源，运行结束后，还需要资源释放，下次还需再重新申请资源才能运行。显然，这种任务会频繁启停的情况不适用于 Per Job 模式，更适合用 Session 模式。

#### 实践 ####

如果你只想运行单个 Flink Job 后就退出，那么可以用下面这个命令：

./bin/flink run -m yarn-cluster -yn 2 examples/streaming/WordCount.jar --input hdfs:///test_dir/input_dir/story --output hdfs:///test_dir/output_dir/output

常用的配置有：

*  \-yn,–yarncontainer Number of Task Managers
 *  \-yqu,–yarnqueue Specify YARN queue.
 *  \-ys,–yarnslots Number of slots per TaskManager
 *  \-yqu,–yarnqueue Specify YARN queue.

可以通过 Help 命令查看 Run 的可用参数：

./bin/flink run -h

我们可以看到，“./bin/flink run -h”看到的“Options for yarn-cluster mode”中的“-y”和“–yarn”为前缀的参数其实和“./bin/yarn-session.sh -h”命令是一一对应的，语义上也基本一致。

关于“-n”（在yarn session模式下）、“-yn”在（yarn single job模式下）与“-p”参数的关系：

*  “-n”和“-yn”在社区版本中（Release-1.5 ～ Release-1.7）中没有实际的控制作用，实际的资源是根据“-p”参数来申请的，并且 TM 使用完后就会归还
 *  在 Blink 的开源版本中，“-n”（在 Yarn Session 模式下）的作用就是一开始启动指定数量的 TaskManager，之后即使 Job 需要更多的 Slot，也不会申请新的 TaskManager
 *  在 Blink 的开源版本中，Yarn single job 模式“-yn”表示的是初始 TaskManager 的数量，不设置 TaskManager 的上限。（需要特别注意的是，只有加上“-yd”参数才能用 Single job 模式（例如：命令“./bin/flink run -yd -m yarn-cluster xxx”）

### 参考资料 ###

1.  https://ververica.cn/developers/advanced-tutorial-2-flink-on-yarn-k8s/ Flink on Yarn/K8s 原理剖析及实践
2.  https://www.jianshu.com/p/1b05202c4fb6 Flink on yarn部署模式

[img]: /images/20230528/5be949b367c74528812c1e12ded6ba4d.png
[img 1]: /images/20230528/a7fe7fca1b564b7693b7c3729877ab70.png
[img 2]: /images/20230528/2d35c7fb1ab340ebb3b9cc748cf83748.png