Flink是一个分布式系统，要求有效地分配和管理计算资源以执行流式应用程序。它集成了所有常见的集群资源管理器，如Hadoop YARN和Kubernetes，但也可以设置为作为standalone甚至库运行。

本节概述了Flink的体系结构，并描述了其主要组件如何交互以执行应用程序以及从故障中恢复。

Flink集群解析

Flink运行时由两种类型的进程组成：一个JobManager和一个或多个TaskManager。

Client 不是运行时和程序执行的一部分，而是用于准备数据流并将其发送到JobManager。之后，Client 可以断开连接（分离模式），或者保持连接以接收进度报告（附加模式）。Client 要么作为触发执行的Java/Scala程序的一部分运行，要么在命令行进程/bin/flink run ...中运行

JobManager和TaskManager可以通过各种方式启动：直接在机器上作为standalone启动，在容器中启动，或者由YARN等资源框架管理。TaskManager连接到JobManager，宣布自己可用，并被分配工作。

JobManager

JobManager 有许多与协调Flink应用程序的分布式执行相关的职责：它决定何时安排下一个任务（或一组任务），对已完成或执行失败的任务做出反应，协调检查点，并协调故障恢复等。该进程由三个不同的组件组成：

• ResourceManager

  ResourceManager 负责Flink 集群中的资源分配和供应，管理任务槽(task slots) --是Flink集群的资源调度单元。Flink为不同的环境和资源提供商（如YARN、Kubernetes和独立部署）实现了多个ResourceManager。在standalone设置中，ResourceManager 只能分配可用TaskManager的插槽，不能独立启动新的TaskManager。

• Dispatcher

  Dispatcher提供了一个REST接口来提交Flink应用程序以供执行，并为每个提交的Job启动一个新的JobMaster。同时，Dispatcher还运行Flink WebUI提供job执行信息

• JobMaster

  JobMaster负责管理单个JobGraph的执行。一个Flink cluster中可以同时运行多个job，每个job都有自己的JobMaster。

至少有一个JobManager。一个高可用性设置可能有多个JobManager，其中一个始终是leader，其他则是备用(standby)（请参阅高可用性(HA))。

TaskManager

TaskManager（也称为worker）执行数据流任务，缓冲和交换数据流。

必须始终至少有一个TaskManager。TaskManager中资源调度的最小单位是任务槽(task slot)。任务槽的数量表示并发处理任务的数量。请注意，可能在一个任务槽中执行多个Operator

Task和算子(Operator)链

对于分布式执行，Flink 将算子的 subtasks 链接成 tasks。每个task由一个线程执行。将将operator链接成task是一种有用的优化：它减少了线程切换和缓冲的开销，并在降低延迟的同时提高了整体吞吐量。可以配置链接行为；请参阅chaining docs查看详细信息。

下图中的示例数据流由五个Subtask执行，因此由五个并行线程执行

Task Slot(任务槽)和资源

每个worker(TaskManager)都是一个JVM进程，可以在单独的线程中执行一个或多个子任务。为了控制单个TaskManager接受的任务数，就有了所谓的task slot（至少一个）。

每个 task slot 表示TaskManager的固定资源子集。例如，具有三个slot 的TaskManager会将其托管内存的1/3专用于每个插槽。划分资源意味着subtask不会与其他作业的subtask争夺托管内存，而是有一定数量的保留托管内存。请注意，这里没有进行CPU隔离；当前slot仅隔离任务的托管内存。

通过调整task slot 的数量，用户可以定义如何将subtask彼此隔离。每个TaskManager有一个slot 意味着每个任务组都在一个单独的JVM中运行（例如，可以在一个独立的容器中启动）。拥有多个slot 意味着更多的subtask共享同一JVM。同一JVM中的任务共享TCP连接（通过多路复用）和心跳消息。它们还可以共享数据集和数据结构，从而减少每个任务的开销。

默认情况下，Flink允许subtask共享slot ，即使它们是不同task的subtask ，只要来自同一job即可。结果就是，一个slot可以容纳job的整个管道。允许这种“slot共享”有两个主要好处：

• Flink集群所需task slot与job使用的最大并行度保持一样。不需要计算一个程序总共包含多少任务（具有不同的并行度）。
• 更容易获得更好的资源利用率。如果没有“slot共享”，非密集型subtask(source/map()) 将阻塞与资源密集型 subtask(window）一样多的资源。通过“slot共享”，将示例中的基本并行度从两个增加到六个，可以充分利用slot资源，同时确保繁重的subtask在TaskManager之间公平分配。

Flink 应用程序执行

• 集群生命周期: Flink应用集群是一个专用的Flink集群，它只执行来自一个Flink应用的job，并且 main() 方法在集群上运行，而不是在client运行。job提交是一个一步到位的过程: 你不需要先启动Flink集群，然后向现有集群会话提交job ，相反，你将应用程序逻辑和依赖项打包到一个可执行的作业JAR包中，集群入口点(ApplicationClusterEntryPoint) 负责调用main() 方法来提取JobGraph。这允许你像Kubernetes上的任何其他应用程序一样部署Flink应用程序。Flink应用程序集群的生命周期因此与Flink应用的生命周期绑定。
• 资源隔离: 在Flink应用集群中，ResourceManager和Dispatcher的作用域为一个Flink应用，它提供了比Flink会话集群更好的隔离。

Flink Session集群

• 集群生命周期: 在Flink会话集群中，客户端连接到一个预先存在的、长期运行的集群，该集群可以接受多个job提交。即使在所有job完成后，集群(和JobManager) 仍将继续运行，直到手动停止会话。因此，Flink会话集群的生存期不与任何Flink job的生存期绑定。
• 资源隔离: TaskManager slot由ResourceManager在提交job时分配，并在job完成后释放。因为所有作业都共享同一个集群，所以在提交job阶段存在一些集群资源竞争，比如网络带宽。这种共享设置的一个限制是，如果一个TaskManager崩溃，那么所有在该TaskManager上运行任务的job都将失败；类似的，如果JobManager上发生一些致命错误，它将影响集群中运行的所有job。
• 其他注意事项: 拥有预先存在的集群可以节省大量申请资源和启动TaskManager的时间。在job的执行时间非常短，且启动时间过长会对端到端用户体验产生负面影响的情况下，这一点很重要——短查询的交互式分析就是这样，希望job可以使用现有资源快速执行计算。

以前，Flink会话集群也称为session mode下的Flink集群。

参考链接

https://nightlies.apache.org/flink/flink-docs-master/docs/concepts/flink-architecture/