Flink 架构

前言

Flink 是一个多功能框架，能够以组合形式适配许多不同的部署场景。
在接下来的文章中，我们大致介绍了 Flink 集群的组成部分，它们的作用以及不同的实现方法。

1、架构总览

下图展示了 Flink 集群的组成架构。

Flink 架构总览

• client：处理 Flink 应用代码，将代码转换成 JobGraph 形式然后提交给 JobManager
• JobManager：分发工作，将各 Job 的任务分发给各个 TaskManager
• TaskManager：真正处理工作的组件，运行 sources、transformations 以及 sinks 等算子

在实际部署 Flink 的过程中，通常每个组件都有多种实现，我们会在后文中进行介绍。

2、Flink 固有组件

2.1 Flink Client

2.1.1 作用

• 将批处理或者流处理应用编译成 dataflow graph
• 将 JobGraph 提交给 JobManager

2.1.2 实现

• Command Line Interface
• REST Endpoint
• SQL Client
• Python REPL
• Scala REPL

2.2 JobManager

2.2.1 作用

JobManager 被称为 Flink 的中心工作调度组件。根据资源提供商的不同，在 HA、资源分配以及不同作业提交模式的支持性方面均有不同的实现方法。
相关进程从功能上可以分为以下3类：

• ResourceManager：负责 Flink 集群中的资源（即：task slots，集群资源调度的基本单元）调配工作。仅能调度已有的 slots，无法自己启动新的 TaskManagers。
• Dispatcher：提供一个 REST 风格的接口用以提交 Flink 应用的执行请求。为每个提交过来的 job 启动一个 JobMaster 进程。同时提供 Flink WebUI 用以提供集群中 job 的执行情况。
• JobMaster：负责管理单个 JobGraph 的执行过程。同一个集群中可以同时运行多个 job，每个 job 都有自己的 JobMaster。

Flink 集群中至少会有一个 JobManager，如果开启了高可用模式，则会有多个，其中一个是 leader，其他的是 standby 状态。

根据不同的作业提交模式，JobManager 可以分为以下三种模式：

• Application Mode：单集群只跑单应用。job 的 main 方法在 JobManagers 上执行，且支持多次调用应用的 execute/executeAsync 接口
• Per-Job Mode：单集群只跑单 job。job 的 main 方法在 client 端执行
• Session Mode：一个集群运行多个应用，各应用的 jobs 共享集群资源

2.2.2 实现

• Standalone (this is the barebone mode that requires just JVMs to be launched. Deployment with Docker, Docker Swarm / Compose, non-native Kubernetes and other models is possible through manual setup in this mode)
• Kubernetes
• YARN
• Mesos

2.3 TaskManager

2.3.1 作用

TaskManager 又被称为 workers，每个 TaskManager 其实就是一个 JVM 进程，通过不同线程处理不同的子任务，对数据流中的数据执行各种算子、缓存并交换数据流。
集群中最少要有一个 TaskManager，TaskManager 中最小的调度单元被称为 task slot，TaskManager 中的 task slot 数量体现了可以同时处理 task 的数量，默认情况下一个 TaskManager 进程有一个 job slot，通过配置 taskmanager.numberOfTaskSlots可以设置更多的 job slot。注意，同一个 task slot 中可能会执行多个算子。

关于 Tasks 与 算子链
为实现分布式执行，Flink 会将算子的 subtasks 链接成 task，每个 task 都用一个线程来执行。这是一个很有用的策略：它减少了线程之间切换和缓冲的开销，在降低延迟的同时提高了整体吞吐量。链接行为是可以配置的，详情点此。

3、Flink 外部组件

3.1 HA 服务提供商

Flink 的 JobManager 组件可以以 HA 模式运行，该模式允许 Flink 能够从 JobManager 故障中自动恢复。为了保证更快的故障恢复，会启动多个备份 JobManager 实例，随时做好准备。
目前已有下面两种实现：

• Zookeeper
• Kubernetes HA

3.2 文件存储及持久性实现

Flink 依赖外部文件存储系统来实现其 checkpoint 机制，即：流式作业的恢复机制，详见 FileSystems。

3.3 资源提供商

Flink 可以部署到多种资源提供商平台，如：Kubernetes、YARN 以及 Mesos。
各平台具体实现参考 JobManager。

3.4 指标存储

Flink 各组件报告指标，且 Flink job 还会额外报告 job 本身特有的指标。
详见 Metrics Reporter。

3.5 应用数据 Sources 以及 Sinks

Sources：数据来源，如文件、目录、sockets 等
Sinks：Flink 生成数据去向，如写入文件、打印到标准输出或者 sockets 等

严格意义上来说，在部署 Flink 集群时，应用数据的 Sources 以及 Sinks 并不属于任何组件，而是在设计新 Flink 生产集群之初就要考虑进去的。将常用数据与 Flink 合并可以获得显著的性能优势。

3.5.1 常用的 Sources 以及 Sinks：

• Apache Kafka (source/sink)
• Apache Cassandra (sink)
• Amazon Kinesis Streams (source/sink)
• Elasticsearch (sink)
• FileSystem (Hadoop included) - Streaming only (sink)
• FileSystem (Hadoop included) - Streaming and Batch (sink)
• RabbitMQ (source/sink)
• Apache NiFi (source/sink)
• Twitter Streaming API (source)
• Google PubSub (source/sink)
• JDBC (sink)

详见 Connectors。

4、部署模式

Flink 可以通过以下三种方法来运行应用：

• Application Mode
• Per-Job Mode
• Session Mode

不同之处在于：

• 集群生命周期以及资源隔离保障
• 应用main方法的执行位置：在 client 端还是在集群中

Flink 部署模式

4.1 Application Mode

在其他运行模式中，应用的main方法都是在 client 端执行的，执行过程包括本地下载应用的依赖、执行main方法并将应用转换成 Flink 运行时可以理解的形式（如： JobGraph），然后将依赖以及 JobGraph 提交给集群。这个过程需要大量的宽带来完成依赖下载以及二进制文件的提交工作，同时需要 CPU 来执行main方法，使得 Client 成为了一个资源消耗大头，当 Client 被多个用户共享时，这个问题将变得尤其明显。

基于上述问题的考虑，Application Mode 采取了不同的处理机制，在此模式下，Flink 会为每个应用创建一个集群，并且选择在 JobManager 上运行该应用的main方法。新创建的集群资源，只会被同一个应用的 job 共享使用，并且当应用执行结束时，集群也会随之被释放。通过这种架构，用户能够像 Per-Job Mode 一样实现资源隔离以及负载均衡，当然是从应用的粒度来说。

把应用main方法的执行过程放到 JobManager 上可以节省 CPU，也减少了本地下载依赖的宽带开销。同时，由于本应用所有 Job 都共享一个 JobManager，集群在下载应用程序的依赖项时，可以更均匀地平衡网络负载。

注意：由于 Application Mode 将main方法的执行过程安排在集群里而不是 client 端，这可能会对应用的代码有影响。举个例子，代码中使用registerCachedFile()方法注册的所有路径，都必须能够被本应用所属集群的 JobManager 访问。

相比于 Per-Job Mode，Application Mode 允许 client 提交包含多个 jobs 的应用。应用中各 job 的执行顺序由该 job 的触发方式决定。如果使用execute()方法触发 job，执行过程会阻塞，只有上一个 job 完成后，下一个 job 才会被执行到。如果使用executeAsync()方法触发 job，执行过程不会阻塞，不管上一个 job 是否已完成，这个 job 都会开始执行。

注意：当应用中包含多个 job 时，集群将不支持 HA

4.2 Per-Job Mode

Per-Job Mode 利用 YARN、Kubernetes 这样的资源提供框架为每个 client 提交的 job 创建一个集群，以实现更好的资源隔离机制。创建的集群，仅限该 job 使用，当 job 执行完成后，集群就会被释放，包括所有的资源（文件等）。这种情况下，job 执行异常时，破坏的也仅仅是它自己的 TaskManagers，可以带来更好的资源隔离体验。除此之外，由于每个 job 的集群是独立的，JobManagers 之间的 book-keeping 负载开销也更均衡。也正是由于这些原因，Per-Job 模式下的资源分配模型更加受许多生产环境实施的喜爱。

4.3 Session Mode

Session Mode 假设已经有一个运行的集群，并且使用该集群的资源来执行所有提交的应用。
集群中应用共享集群资源，并因此产生竞态，这样做的好处是，你不必费力为每个提交的 job 创建集群，但是，如果某个 job 执行异常或者损坏了某个 TaskManager，该 TaskManager 上运行的其他所有 job 也会受影响，因此除了那个执行异常的 job 受影响之外，通常还意味着出现大规模的恢复过程，所有重启的 job 都会读取文件系统，造成其他服务无法访问。除此之外，在单集群中运行多个 job 会给 JobManager 带来更高的负载，因为它需要 book-keeping 集群中所有的 job。

4.4 总结

Session Mode 模式 Flink 集群的生命周期跟集群中运行的 job 无关，是独立运行的，集群资源被所有 job 共享。
Per-Job 模式为了提供更好的资源隔离体验，牺牲精力为每个提交的 job 创建集群，因此，集群的生命周期是与一个个的 job 绑定在一起的。
Application 模式为每个应用创建一个会话集群，并且在集群里执行该应用的main方法。