flink-1.11 Application 模式

1. 背景

flink-1.11 引入了一种新的部署模式，即 Application 模式。目前，flink-1.11 已经可以支持基于 Yarn 和 Kubernetes 的 Application 模式。

2. 优势

Session模式：所有作业共享集群资源，隔离性差，JM 负载瓶颈，main 方法在客户端执行。
Per-Job模式：每个作业单独启动集群，隔离性好，JM 负载均衡，main 方法在客户端执行。

通过以上两种模式的特点描述，可以看出，main方法都是在客户端执行，社区考虑到在客户端执行 main() 方法来获取 flink 运行时所需的依赖项，并生成 JobGraph，并将依赖项和 JobGraph 发送到集群的一系列过程中，由于需要大量的网络带宽下载依赖项并将二进制文件发送到集群，会造成客户端消耗大量的资源。尤其在大量用户共享客户端时，问题更加突出。因此，社区提出新的部署方式 Application 模式解决该问题。

3. 原理

Application 模式下，用户程序的 main 方法将在集群中而不是客户端运行，用户将程序逻辑和依赖打包进一个可执行的 jar 包里，集群的入口程序 (ApplicationClusterEntryPoint) 负责调用其中的 main 方法来生成 JobGraph。Application 模式为每个提交的应用程序创建一个集群，该集群可以看作是在特定应用程序的作业之间共享的会话集群，并在应用程序完成时终止。在这种体系结构中，Application 模式在不同应用之间提供了资源隔离和负载平衡保证。在特定一个应用程序上，JobManager 执行 main() 可以节省所需的 CPU 周期，还可以节省本地下载依赖项所需的带宽。

4. 使用

application 模式 使用 bin/flink run-application 提交作业；通过 -t 指定部署环境，目前 application 模式支持部署在 yarn 上(-t yarn-application) 和 k8s 上(-t kubernetes-application；并支持通过 -D 参数指定通用的 运行配置，比如 jobmanager/taskmanager 内存、checkpoint 时间间隔等。
通过 bin/flink run-application -h 可以看到 -D/-t 的详细说明：(-e 已经被废弃，可以忽略)

Options for Generic CLI mode:
     -D <property=value>   Generic configuration options for
                           execution/deployment and for the configured executor.
                           The available options can be found at
                           https://ci.apache.org/projects/flink/flink-docs-stabl
                           e/ops/config.html
     -e,--executor <arg>   DEPRECATED: Please use the -t option instead which is
                           also available with the "Application Mode".
                           The name of the executor to be used for executing the
                           given job, which is equivalent to the
                           "execution.target" config option. The currently
                           available executors are: "collection", "remote",
                           "local", "kubernetes-session", "yarn-per-job",
                           "yarn-session".
     -t,--target <arg>     The deployment target for the given application,
                           which is equivalent to the "execution.target" config
                           option. The currently available targets are:
                           "collection", "remote", "local",
                           "kubernetes-session", "yarn-per-job", "yarn-session",
                           "yarn-application" and "kubernetes-application".

下面列举几个使用 Application 模式提交作业到 yarn 上运行的命令：

1.带有 JM 和 TM 内存设置的命令提交：

./bin/flink run-application -t yarn-application \
-Djobmanager.memory.process.size=2048m \
-Dtaskmanager.memory.process.size=4096m \
-Dyarn.application.name="MyFlinkWordCount" \
./examples/batch/WordCount.jar

2.在 1 的基础上自己设置 TaskManager slots 个数为3，以及指定并发数为3：

./bin/flink run-application -t yarn-application -p 3 \
-Djobmanager.memory.process.size=2048m \
-Dtaskmanager.memory.process.size=4096m \
-Dyarn.application.name="MyFlinkWordCount" \
-Dtaskmanager.numberOfTaskSlots=3 \
./examples/batch/WordCount.jar

当然，指定并发还可以使用 -Dparallelism.default=3，而且社区目前倾向使用 -D+通用配置代替客户端命令参数(比如 -p)。
所以这样写更符合规范：

./bin/flink run-application -t yarn-application -p 3 \
-Dparallelism.default=3 \
-Djobmanager.memory.process.size=2048m \
-Dtaskmanager.memory.process.size=4096m \
-Dyarn.application.name="MyFlinkWordCount" \
-Dtaskmanager.numberOfTaskSlots=3 \
./examples/batch/WordCount.jar

3.和 yarn.provided.lib.dirs 参数一起使用，可以充分发挥 application 部署模式的优势：
我们看 官方配置文档 对这个配置的解释：

yarn.provided.lib.dirs: A semicolon-separated list of provided lib directories. They should be pre-uploaded and world-readable. Flink will use them to exclude the local Flink jars(e.g. flink-dist, lib/, plugins/)uploading to accelerate the job submission process. Also YARN will cache them on the nodes so that they doesn't need to be downloaded every time for each application. An example could be hdfs://$namenode_address/path/of/flink/lib

意思是我们可以预先上传 flink 客户端依赖包 (flink-dist/lib/plugin) 到远端存储(一般是 hdfs，或者共享存储)，然后通过 yarn.provided.lib.dirs 参数指定这个路径，flink 检测到这个配置时，就会从该地址拉取 flink 运行需要的依赖包，省去了依赖包上传的过程，yarn-cluster/per-job 模式也支持该配置。在之前的版本中，使用 yarn-cluster/per-job 模式，每个作业都会单独上传 flink 依赖包(一般会有 180MB左右)导致 hdfs 资源浪费，而且程序异常退出时，上传的 flink 依赖包往往得不到自动清理。通过指定 yarn.provided.lib.dirs，所有作业都会使用一份远端 flink 依赖包，并且每个 yarn nodemanager 都会缓存一份，提交速度也会大大提升，对于跨机房提交作业会有很大的优化。
使用示例如下：
my-application.jar 是用户 jar 包

./bin/flink run-application -t yarn-application \
-Djobmanager.memory.process.size=2048m \
-Dtaskmanager.memory.process.size=4096m \
-Dyarn.application.name="MyFlinkWordCount" \
-Dtaskmanager.numberOfTaskSlots=3 \
-Dyarn.provided.lib.dirs="hdfs://myhdfs/my-remote-flink-dist/lib;hdfs://myhdfs/my-remote-flink-dist/plugins" \
/local/path/to/my-application.jar

my-application.jar 也可以提前上传 hdfs：

./bin/flink run-application -t yarn-application \
-Djobmanager.memory.process.size=2048m \
-Dtaskmanager.memory.process.size=4096m \
-Dyarn.application.name="MyFlinkWordCount" \
-Dtaskmanager.numberOfTaskSlots=3 \
-Dyarn.provided.lib.dirs="hdfs://myhdfs/my-remote-flink-dist/lib;hdfs://myhdfs/my-remote-flink-dist/plugins" hdfs://myhdfs/jars/my-application.jar

也可以将 yarn.provided.lib.dirs 配置到 conf/flink-conf.yaml，这时提交作业就和普通作业没有区别了：

./bin/flink run-application -t yarn-application \
-Djobmanager.memory.process.size=2048m \
-Dtaskmanager.memory.process.size=4096m \
-Dyarn.application.name="MyFlinkWordCount" \ 
-Dtaskmanager.numberOfTaskSlots=3 \
/local/path/to/my-application.jar

注意：如果自己指定 yarn.provided.lib.dirs，有以下注意事项：

• 需要将 lib 包和 plugins 包地址用;分开，从上面的例子中也可以看到，将 plugins 包放在 lib 目录下可能会有包冲突错误
• plugins 包路径地址必须以 plugins 结尾，例如上面例子中的 hdfs://myhdfs/my-remote-flink-dist/plugins
• hdfs 路径必须指定 nameservice(或 active namenode 地址)，而不能使用简化方式(例如 hdfs:///path/to/lib)

该种模式的操作使得 flink 作业提交变得很轻量，因为所需的 Flink jar 包和应用程序 jar 将到指定的远程位置获取，而不是由客户端下载再发送到集群。这也是社区在 flink-1.11 版本引入新的部署模式的意义所在。

Application 模式在停止、取消或查询正在运行的应用程序的状态等方面和 flink-1.11 之前的版本一样，可以采用现有的方法。