Flink 是一个用于状态化并行流处理的分布式系统。它的搭建涉及多个进程，这些进程通常会分布在多台机器上。分布式系统需要应对的常见挑战包括 分配和管理集群计算资源，进程协调，持久且高可用的数据存储 及 故障恢复 等。

Flink 并没有依靠自身实现所有上述功能，而是在已有集群基础设施和服务之上专注于它的核心功能：分布式数据流处理。Flink 和很多集群管理器（如 Apache Mesos、YARN 及 Kubernetes）都能很好地集成；同时它也可以通过配置，作为独立集群来运行。Flink 没有提供分布式持久化存储，而是利用了现有的分布式文件系统（如 HDFS）或对象存储（如 S3）。它依赖 Apache ZooKeeper 来完成高可用性设置中的领导选举工作。

本篇博客我们将介绍搭建 Flink 时所涉及的不同组件并讨论它们在应用运行时的交互过程。我们主要讨论两类部署 Flink 应用的方式以及它们如何分配和执行任务。最后，我们将解释 Flink 高可用模式的工作原理。

1.Flink 组件

Flink 的搭建需要四个不同组件，它们相互协作，共同执行流式应用。这些组件是：JobManager、ResourceManager、TaskManager 和 Dispatcher。Flink 本身是用 Java 和 Scala 实现的，因此所有组件都基于 Java 虚拟机（JVM）运行。

1.1 JobManager

作为主进程（master process），JobManager 控制着单个应用程序的执行。换句话说，每个应用都由一个不同的 JobManager 掌控。JobManager 可以接收需要执行的应用，该应用会包含一个所谓的 JobGraph，即 逻辑 Dataflow 图，以及一个打包了全部所需类、库以及其他资源的 JAR 文件。JobManager 将 JobGraph 转化成名为 ExecutionGraph 的 物理 Dataflow 图，该图包含了那些可以并行执行的任务。JobManager 从 ResourceManager 申请执行任务的必要资源（TaskManager 处理槽）。一旦它收到了足够数量的 TaskManager 处理槽（slot），就会将 ExecutionGraph 中的任务分发给 TaskManager 来执行。在执行过程中，JobManager 还要负责所有需要集中协调的操作，如创建检查点。

1.2 ResourceManager

针对不同的环境和资源提供者（resource provider）（如 YARN、Mesos、Kubernetes 或独立部署），Flink 提供了不同的ResourceManager。ResourceManager 负责管理 Flink 的处理资源单元：TaskManager 处理槽。当 JobManager 申请 TaskManager 处理槽时，ResourceManager 会指示一个拥有空闲处理槽的 TaskManager 将其处理槽提供给 JobManager。如果 ResourceManager 的处理槽数无法满足 JobManager 的请求，则 ResourceManager 可以和资源提供者通信，让它们提供额外容器来启动更多 TaskManager 进程。同时，ResourceManager 还负责终止空闲的 TaskManager 以释放计算资源。

1.3 TaskManager

TaskManager 是 Flink 的 工作进程（worker process）。通常在 Flink 搭建过程中要启动多个 TaskManager。每个 TaskManager 提供一定数量的处理槽。处理槽的数目限制了一个 TaskManager 可执行的任务数。TaskManager 在启动后，会向 ResourceManager 注册 它的处理槽。当接收到 ResourceManager 的指示时，TaskManager 会向 JobManager 提供一个或多个处理槽。之后，JobManager 就可以向处理槽中分配任务来执行。在执行期间，运行同一应用不同任务的 TaskManager 之间会产生数据交换。我们将在稍后进一步讨论任务执行和处理槽的概念。

1.4 Dispatcher

Dispatcher 会跨多个作业运行，它提供了一个 REST 接口来让我们提交需要执行的应用。一旦某个应用提交执行，Dispatcher 会启动一个 JobManager 并将应用转交给它。REST 接口意味着 Dispatcher 这一集群的 HTTP 入口可以受到防火墙的保护。Dispatcher 同时还会启动一个 WebUI，用来提供有关作业执行的信息。某些应用提交执行的方式可能用不到 Dispatcher。

下图展示了应用提交执行过程中 Flink 各组件之间的交互过程。

上图仅是一个用于展示各组件职责与交互的高层次框架。根据环境的不同（YARN、Mesos、Kubernetes 或独立集群），图中某些步骤其实可以省略，同时也可会有多个组件运行在同一 JVM 进程中。例如：独立集群设置下没有资源提供者，因此 ResourceManager 只能分配现有 TaskManager 中的处理槽而无法自己启动新的 TaskManager。

2.应用部署

Flink应用可以通过两种模式进行部署。

2.1 框架模式

在该模式下，Flink 应用会打包成一个 JAR 文件，通过客户端提交到运行的服务上。这里的服务可以是 Flink Dispatcher，Flink JobManager 或是 YARN 的 ResourceManager。无论哪种情况，运行的服务都会接收 Flink 应用并确保其执行。如果应用提交到 JobManager，会立即开始执行；如果应用提交到 Dispatcher 或 YARN ResourceManager，它们会启动一个 JobManager 并将应用转交给它，随后由 JobManager 负责执行该应用。

2.2 库模式

在该模式下，Flink 应用会绑定到一个特定应用的容器镜像（如 Docker 镜像）中。镜像中还包含着运行 JobManager 以及 ResourceManager 的代码。当容器从镜像启动后会自动加载 ResourceManager 和 JobManager，并将绑定的作业提交执行。另一个和作业无关的镜像负责部署 TaskManager 容器。容器通过镜像启动后会自动运行 TaskManager，后者可以连接 ResourceManager 并注册处理槽。通常情况下，外部资源管理框架（如 Kubernetes）负责启动镜像，并确保在发生故障时容器能够重启。

基于框架的模式采用的是传统方式，即通过客户端提交应用（或查询）到正在运行的服务上；而在库模式中，Flink不是作为服务，而是以库的形式绑定到应用所在的容器镜像中。后者常用于微服务架构。我们会在第10章的“运行和管理流式应用”中详细讨论应用部署的相关内容。

3.任务执行

一个 TaskManager 允许同时执行多个任务。这些任务可以属于同一个算子（数据并行），也可以是不同算子（任务并行），甚至还可以来自不同的应用（作业并行）。TaskManager 通过提供固定数量的处理槽来控制可以并行执行的任务数。每个处理槽可以执行应用的一部分，即算子的一个并行任务。下图展示了 TaskManager、处理槽、任务以及算子之间的关系。

左侧的 JobGraph（应用的非并行化表示）包含了 5 个算子，其中算子 A 和 C 是数据源，算子 E 是数据汇。算子 C 和 E 的并行度为 2，其余算子的并行度为 4。由于算子最大并行度是 4，因此应用若要执行则至少需要 4 个处理槽。如果每个 TaskManager 内有两个处理槽，则运行两个 TaskManager 即可满足该需求。JobManager 将 JobGraph “展开成” ExecutionGraph 并把任务分配到 4 个空闲处理槽。对于并行度为 4 的算子，其任务会每个处理槽分配一个。其余两个算子 C 和 E 的任务会分别放到处理槽 1.1、2.1 和处理槽 1.2、2.2 中。将任务以切片的形式调度至处理槽中有一个好处：TaskManager 中的多个任务可以在同一进程内高效地执行数据交换而无须访问网络。然而，任务过于集中也会使 TaskManager 负载变高，继而可能导致性能下降。

TaskManager 会在同一个 JVM 进程内以多线程的方式执行任务。和独立进程相比，线程更加轻量并且通信开销更低，但无法严格地将任务彼此隔离。因此只要有一个任务运行异常，就有可能 “杀死” 整个 TaskManager 进程，导致它上面运行的所有任务都停止。如果将每个 TaskManager 配置成只有一个处理槽，则可以限制应用在 TaskManager 级别进行隔离，即每个 TaskManager 只运行单个应用的任务。通过在 TaskManager 内部采用线程并行以及在每个主机上部署多个 TaskManager 进程，Flink 为部署应用时性能和资源隔离的取舍提供了极大的自由度。我们会在后续讨论搭建和配置 Flink 集群的详细内容。

4.高可用设置

流式应用通常都会设计成 7 x 24 小时运行，因此对于它很重要的一点是：即便内部进程发生故障时也不能终止运行。为了从故障中恢复，系统首先要 重启故障进程，随后需要 重启应用并恢复其状态。接下来我们将介绍 Flink 如何重启故障进程。而恢复应用状态则会在后续有关 “检查点” 的博客中进行介绍。

4.1 TaskManager 故障

如前所述，为了执行应用的全部任务，Flink 需要足够数量的处理槽。假设一个 Flink 设置包含 4 个 TaskManager，每个 TaskManager 有 2 个处理槽，那么一个流式应用最多支持以并行度 8 来运行。如果有一个 TaskManager 出现故障，则可用处理槽的数量就降到了 6 个。这时候 JobManager 就会向 ResourceManager 申请更多的处理槽。若无法完成（例如应用运行在一个独立集群上），JobManager 将无法重启应用，直至有足够数量的可用处理槽。应用的 重启策略 决定了 JobManager 以何种频率重启应用以及重启尝试之间的等待间隔。

4.2 JobManager 故障

和 TaskManager 相比，JobManager 发生故障会更为棘手。它用于控制流式应用执行以及保存该过程中的元数据（如已完成检查点的存储路径）。如果负责管理的 JobManager 进程消失，流式应用将无法继续处理数据。这就导致 JobManager 成为 Flink 应用中的一个单点失效组件。为了解决该问题，Flink 提供了高可用模式，支持在原 JobManager 消失的情况下将作业的管理职责及元数据迁移到另一个 JobManager。

Flink 中的高可用模式是基于能够提供分布式协调和共识服务的 Apache ZooKeeper 来完成的，它在 Flink 中主要用于 “领导” 选举以及持久且高可用的数据存储。JobManager 在高可用模式下工作时，会将 JobGraph 以及全部所需的元数据（例如应用的 JAR 文件）写入一个 远程持久化存储系统 中。此外，JobManager 还会将存储位置的路径地址写入 ZooKeeper 的数据存储。在应用执行过程中，JobManager 会接收每个任务检查点的 状态句柄（存储位置）。在检查点即将完成的时候，如果所有任务已经将各自状态成功写入远程存储，JobManager 就会将状态句柄写入远程存储，并将远程位置的路径地址写入 ZooKeeper。因此所有用于 JobManager 故障恢复的数据都在远程存储上面，而 ZooKeeper 持有这些存储位置的路径。

当 JobManager 发生故障时，其下应用的所有任务都会自动取消。新接手工作的 JobManager 会执行以下步骤：

• 1️⃣ 向 ZooKeeper 请求存储位置，以获取 JobGraph、JAR 文件及应用最新检查点在远程存储的状态句柄。
• 2️⃣ 向 ResourceManager 申请处理槽来继续执行应用。
• 3️⃣ 重启应用并利用最近一次检查点重置任务状态。

如果是在容器环境（如 Kubernetes）中以库模式部署运行应用，容器编排服务（orchestration service）通常会自动重启故障的 JobManager 或 TaskManager 容器。当运行在 YARN 或 Mesos 上面时，Flink 的其余进程会触发 JobManager 或 TaskManager 进程重启。Flink 没有针对独立集群模式提供重启故障进程的工具，因此有必要运行一些后备 JobManager 及 TaskManager 来接管故障进程的工作。