Flink 分布式运行时环境

任务与运算链（Operator Chains）

在分布式运行中，Flink 会将多个运算子任务（operator subtasks）在 task 中组成 chains 。每个线程执行一个任务。将运算符链接到计算任务中对于系统性能的提升有很大的帮助：降低了线程间切换与缓冲的开销，在降低延时的同时提高系统的总吞吐量。可以对这种链接（chaining）操作进行配置。

下图所示的数据流图包含五个子任务（汇总视图和并发视图）：

Task chains

Job Manager，Task Manager，Client

Flink 运行时环境由两种类型进程组成：

• JobManager，也称为 master，用于协调分布式执行。负责调度任务，检查点，失败恢复等。

• TaskManager，也称为 worker，用于执行数据流图的任务（更准确地说，是计算子任务），并对数据流进行缓冲、交换。Flink 运行环境中至少包含一个任务管理器。

Program

JobManager 和 TaskManager 可以以多种方式启动：直接作为独立的集群在机器上启动，或者在容器中启动，或者通过 YARN、Mesos 这类资源框架进行管理。启动之后，TaskManagers 会连接 JobManager 报告自身的状态，等待任务分配。

Client 并不是运行时环境的一部分，主要作用是准备并向 JobManager 发送数据流图（dataflow）。在此之后，Client 可以断开连接，也可以保持连接以等待程序运行结果。Client 程序可以是 Java/Scala 程序的形式执行，也可以以命令行的形式（./bin/flink run ...）执行。

Task Slots 和资源

每个 worker（TaskManager）都是一个独立的JVM进程，在独立线程里运行一个或更多的子任务。为了控制 worker 接收任务的数量，在 worker 中引入了任务槽（Task slots）的概念（每个 worker 中至少包含一个 slot）。

每个 Task slot 代表 TaskManager 中一个固定的资源池子集。如果一个 TaskManager 有3个 slots，每个 slot 会分配其 1/3 的内存。将资源进行分槽可以让子任务避免同其他作业中的子任务竞争资源。注意，这里没有对 CPU 进行隔离；目前任务槽仅仅用于划分任务的内存。

通过调整 Task slot 的数量，用户可以设定子任务之间独立运行的程度。如果 TaskManager 中只有一个槽，那么每个任务组都会在一个独立的JVM中运行。TaskManager 中配置更多的槽就意味着会有更多的子任务共享同一个 JVM。在同一个 JVM 中的任务会共享 TCP 连接（通过多路复用的方式）和心跳信息，同时他们也会共享数据集和数据结构，这在某种程度上可以降低单个任务的开销。

Task slots

默认情况下，Flink 会允许同一个作业的多个 subtasks 共享一个 slot，即便这些 subtasks 来自不同的任务。这种情况下，有可能会出现某个 slot 中包含一个完整的作业 pipeline 的情况。这种 slot sharing 主要有两点好处：

• Flink 集群需要在作业中确保 Task slot 数量和程序最高并发量完全一致，并不需要计算程序中任务（每个任务的并发量也许不相同）的具体数量。

• 可以提高资源利用率。如果没有任务槽共享机制（slot sharing），非密集型的 source/map() 子任务就会和密集型的 window 子任务一样阻塞大量资源。如果有任务槽共享机制，在程序的并发量从 2 提高到 6 的情况下（如下图），就可以让密集型子任务完全分散到任务管理器中，从而可以显著提高槽的资源利用率。

Slot sharing

Flink API 中包含一个 resource group 机制，可以避免不合理的任务槽共享。一般来说，默认的任务槽数量应设置为 CPU cores 的数量。

State Backends

存储键值（key/values）索引的数据结构取决于所选的 state backend。有的 state backend 将数据保存在内存中的哈希表中，其他的如使用 RocksDB 会保存 key/value 形式。除了定义保存状态的数据结构之外，state backend 还实现了获取键值对的特定时间点快照的功能，该功能可以将快照保存为检查点的一部分。

State backend

Savepoints

使用 DataStream API 的程序可以从指定的 savepoints 恢复。Savepoints 允许程序和 Flink 集群更新而不丢失任何状态。

Savepoints 可以看作是一种手动触发的检查点，该检查点可以获取程序的快照并将其写入 state backend 中。依赖于一般的检查点机制。执行期间，程序会定期在 worker 节点生成快照和检查点。恢复只需要使用最新一个有效的检查点，一旦新的检查点完成，就可以安全地丢弃旧的检查点。

Savepoints 和定期 checkpoint 在大部分情况下都很相似，区别只在于 Savepoints 是由用户触发的，并且不会自动过期失效。Savepoint 可以通过命令行生成，也可以在调用 REST API 取消作业时产生。

---

Reference
https://ci.apache.org/projects/flink/flink-docs-release-1.6/concepts/runtime.html