在YARN上使用Flink有3种模式：Per-Job模式、Session模式和Application模式。

 

Flink作业运行

https://weread.qq.com/web/reader/1b932790723982d91b9bd8dk66f3299023a66f041e16858

Flink作业在开发完毕之后，需要提交到Flink集群执行。ClientFrontend是入口，触发用户开发的Flink应用Jar文件中的main方法，然后交给PipelineExecutor#execue方法，最终会选择一个触发一个具体的PipelineExecutor执行，过程如图8-1所示。

作业执行可以选择Session和Per-Job模式两种集群：

1）Session模式的集群，一个集群中运行多个作业。

2）Per-Job模式的集群，一个集群只运行一个作业，作业执行完毕则集群销毁。

每种模式适合于不同的场景。不同的运行模式和其适用的场景见表8-1。

表8-1 不同模式的适用场景

根据Flink Client提交作业之后是否可以退出Client进程，提交模式又可分为Detached模式和Attached模式。Detached模式下，Flink Client创建完集群之后，可以退出命令行窗口，集群独立运行。Attached模式下，Flink Client创建完集群后，不能关闭命令行窗口，需要与集群之间维持连接，好处是能够感知集群的退出，集群退出之后有机会做一些资源清理等动作，此处的清理是Flink作业可能占用外部的资源，如在金融行业里，作业占用的加密机连接需要在作业退出时释放等。

1. Session模式

该模式下，作业共享集群资源，作业通过Http协议进行提交。

在Flink 1.10版本中提供了3种会话模式：Yarn会话模式、K8s会话模式、Standalone。Standalone模式比较特别，Flink安装在物理机上，不能像在资源集群上一样，可以随时启动一个新集群，所有的作业共享Standalone集群，本质上就是一种Session模式，所以不支持Per-Job模式。

在Session模式下，Yarn作业提交使用yarn-session.sh脚本，K8s作业提交使用kubernetes-session.sh脚本。两者的具体实现不同，但逻辑是类似的，在启动脚本的时候就会检查是否存在已经启动好的Flink Session模式集群，如果没有，则启动一个Flink Session模式集群，然后在PipelineExecutor中，通过Dispatcher提供的Rest接口提交JobGraph，Dispatcher为每个作业启动一个JobMaster，进入作业执行阶段。

2. Per-Job模式

该模式下，一个作业一个集群，作业之间相互隔离。在Flink 1.10版本中，只有Yarn上实现了Per-Job模式，K8s的Per-Job模式在后续版本中会实现。Per-Job模式下，因为不需要共享集群，所以在PipelineExecutor中执行作业提交的时候，可以创建集群并将JobGraph以及所需要的文件等一同提交给Yarn集群，Yarn集群在容器中启动Flink Master进程（即JobManager进程），进行一系列的初始化动作，初始化完毕之后，从文件系统中获取JobGraph，交给Dispatcher。之后的执行流程与Session模式下的执行流程相同。

 

3 flink1.11 Application Mode

但是现在这些平台遇到一个大问题是部署服务是一个消耗资源比较大的服务，并且很难计算出实际资源限制。比如，如果我们取负载的平均值，则可能导致部署服务的资源真实所需的值远远大于限制值，最坏的情况是在一定时间影响所有的线上应用。但是如果我们将取负载的最大值，又会造成很多不必要的浪费。基于此，Flink 1.11 引入了另外一种部署选项 Application Mode， 该模式允许更加轻量级，可扩展的应用提交进程，将之前客户端的应用部署能力均匀分散到集群的每个节点上。

Flink 中的应用执行

Flink中应用的执行会涉及到三部分：Client，JobManager 和 TaskManagers。Client 负责提交应用到集群，JobManager 负责应用执行期间一些必要的记录工作，TaskManager 负责具体的应用执行。具体的架构图如下：

当前部署模式

在引入Application Mode(Flink1.11) 之前，Flink 支持 Session 和 Per-Job 两种mode，这两种有不同的集群生命周期和资源隔离。

Session 模式

Session 模式假定已经存在一个集群，并任何的提交的应用都在该集群里执行。因此会导致资源的竞争。该模式的优势是你无需为每一个提交的任务花费精力去分解集群。但是，如果Job异常或是TaskManager 宕掉，那么该TaskManager运行的其他Job都会失败。除了影响到任务，也意味着潜在需要更多的恢复操作，重启所有的Job，会并发访问文件系统，会导致该文件系统对其他服务不可用。此外，单集群运行多个Job，意味着JobManager更大的负载。这种模式适合启动延迟非常重要的短期作业。

Per-Job 模式

在Per-Job模式下，集群管理器框架（例如YARN或Kubernetes）用于为每个提交的Job启动一个 Flink 集群。Job完成后，集群将关闭，所有残留的资源（例如文件）也将被清除。此模式可以更好地隔离资源，因为行为异常的Job不会影响任何其他Job。另外，由于每个应用程序都有其自己的JobManager，因此它将记录的负载分散到多个实体中。考虑到前面提到的Session模式的资源隔离问题，Per-Job模式适合长期运行的Job，这些Job可以接受启动延迟的增加以支持弹性。

总而言之，在Session 模式下，集群生命周期独立于集群上运行的任何Job，并且集群上运行的所有Job共享其资源。Per-Job模式选择为每个提交的Job承担拆分集群的费用，以提供更好的资源隔离保证，因为资源不会在Job之间共享。在这种情况下，集群的生命周期将与job的生命周期绑定在一起。

应用提交

Flink 应用的执行包含两个阶段：

• pre-flight: 在main()方法调用之后开始。
• runtime: 一旦用户代码调用 execute() 就会触发该阶段。

main()方法使用Flink的API（DataStream API，Table API，DataSet API）之一构造用户程序。当main()方法调用env.execute()时，用户定义的pipeline将转换为Flink运行时可以理解的形式，称为job graph，并将其传送到集群中。

尽管有一些不同，但是 对于 Session 模式 和 Per-Job模式 ， pre-flight 阶段都是在客户端完成的。

对于那些在自己本地计算机上提交任务的场景(本地计算机包含了所有运行Job所需的依赖)，这通常不是问题。但是，对于通过诸如部署服务之类的远程进行提交的场景，此过程包括：

• 下载应用所需的依赖
• 执行main()方法提取 job graph
• 将依赖和 job graph 传输到集群
• 有可能需要等待结果

这样客户端大量消耗资源，因为它可能需要大量的网络带宽来下载依赖项并将二进制文件运送到集群，并且需要CPU周期来执行main()方法。随着更多用户共享同一客户端，此问题会更加明显。

红色，蓝色和绿色代表3个应用程序，每个应用程序三个并发。黑色矩形代表不同的进程：TaskManagers，JobManagers和 Deployer(集中式部署服务)。并且我们假设在所有情况下都只有一个Deployer进程。彩色三角形表示提交进程的负载，而彩色矩形表示TaskManager和JobManager进程的负载。如图所示，不管是per-job 还是 session 模式， 部署程序承担相同的负载。它们的区别在于Job的分配和JobManager的负载。在session模式下，集群中的所有作业只有一个JobManager，而在per-job模式下，每个Job都有一个JobManager。另外，在session 模式下的Job 被随机分配给TaskManager，而在per-job 模式下，每个TaskManager只有单个Job。

 

Application Mode

Application 模式 尝试去将per-job 模式的资源隔离性和轻量级，可扩展的应用提交进程相结合。为了实现这个目的，它会每个Job 创建一个集群，但是 应用的main()将被在JobManager 执行。

每个应用程序创建一个集群，可以看作创建仅在特定应用程序的Job之间共享的session集群，并在应用程序完成时销毁。通过这种架构，Application模式可以提供与 per-job 模式相同的资源隔离和负载平衡保证，但前提是保证一个完整应用程序的粒度。显然，属于同一应用程序的Job应该被关联起来，并视为一个单元。

在JobManager 中执行 main()方法，更大大减轻客户端的资源消耗。更进一步讲，由于每个应用程序有一个JobManager，因此可以更平均地分散网络负载。上图对此进行了说明，在该图中，这次客户端负载已转移到每个应用程序的JobManager。

在Application 模式下，与其他模式不一样的是，main() 方法在集群上而不是在客户端执行。这可能会对您的代码产生影响，例如，您必须使用应用程序的JobManager可以访问使用registerCachedFile()在环境中注册的任何路径。

与per-job 模式相比，Application 模式允许提交由多个Job组成的应用程序。Job执行的顺序不受部署模式的影响，但受启动Job的调用的影响。使用阻塞的 execute()方法，将是一个顺序执行的效果，结果就是"下一个"Job的执行被推迟到“该”Job完成为止。相反，一旦提交当前作业，非阻塞executeAsync()方法将立即继续提交“下一个”Job。

 

减少网络需求

如上所述，通过在JobManager上执行应用程序的main()方法，Application 模式可以节省很多提交应用所需的资源。但是仍有改进的空间。

专注于YARN， 因为社区对于yarn的优化支持更全面。即使使用 Application 模式，仍然需要客户端将用户jar发送到JobManager。此外，对于每个应用程序，客户端都必须将“ flink-dist”路径输送到集群，该目录包含框架本身的二进制文件，包括flink-dist.jar，lib/ 和plugin/ 目录。这两个可以占用客户端大量的带宽。此外，在每个提交中传送相同的flink-dist二进制文件不仅浪费带宽，而且浪费存储空间，只需允许应用程序共享相同的二进制文件就可以缓解。

对于Flink1.11 , 引入了下面的两个选项可供大家使用：

1. 指定目录的远程路径，YARN可以在该目录中找到Flink分发二进制文件
2. 指定YARN可以在其中找到用户jar的远程路径。

对于1.，我们利用YARN的分布式缓存，并允许应用程序共享这些二进制文件。因此，如果由于先前在同一TaskManager上执行的应用程序而导致某个应用程序恰巧在其TaskManager的本地存储上找到Flink的副本，则它甚至不必在内部下载它。

注意两种优化都可用于YARN上的所有部署模式，而不仅仅是Application模式。

 

示例: Application 模式 on Yarn

有关完整说明，请参阅正式的Flink文档，尤其是涉及集群管理框架，例如YARN或Kubernetes。在这里，我们将提供有关YARN的一些示例:

Application 模式下，使用以下语句提交一个应用：

./bin/flink run-application -t yarn-application ./MyApplication.jar

使用此命令，所有配置参数都可以通过其配置选项（以-D为前缀）来指定。有关可用配置选项的目录，请参阅Flink的配置页面。

例如，用于指定JobManager和TaskManager的内存大小的命令如下所示：

./bin/flink run-application -t yarn-application \
    -Djobmanager.memory.process.size=2048m \
    -Dtaskmanager.memory.process.size=4096m \
    ./MyApplication.jar

为了进一步节省将Flink发行版传送到集群的带宽，请考虑将Flink发行版预上传到YARN可以访问的位置，并使用yarn.provided.lib.dirs配置选项，如下所示：

./bin/flink run-application -t yarn-application \
    -Djobmanager.memory.process.size=2048m \
    -Dtaskmanager.memory.process.size=4096m \
    -Dyarn.provided.lib.dirs="hdfs://myhdfs/remote-flink-dist-dir" \
    ./MyApplication.jar

最后，为了进一步节省提交应用程序jar所需的带宽，您可以将其预上传到HDFS，并指定指向./MyApplication.jar的远程路径，如下所示：

./bin/flink run-application -t yarn-application \
    -Djobmanager.memory.process.size=2048m \
    -Dtaskmanager.memory.process.size=4096m \
    -Dyarn.provided.lib.dirs="hdfs://myhdfs/remote-flink-dist-dir" \
    hdfs://myhdfs/jars/MyApplication.jar

这将使Job提交特别轻巧，因为所需的Flink jar和应用程序jar将从指定的远程位置获取，而不是由客户端传送到集群。客户端将唯一传送到集群的是你的应用程序配置，其中包括上述所有路径。