二、Flink架构

2.1 Flink的设计架构

Flink是一个分层的架构系统，每一层所包含的组件都提供了特定的抽象，用来服务于上层组件，Flink的分层体现有四层，分别是Deploy层、core层、API层/Libraries层。其中Deploy层主要涉及的是Flink的部署模式及同资源调度组件的交互模式，Core层提供了支持Flink计算的全部核心实现，API层/Libraries层提供了Flink的API接口和基于API接口的特定应用的计算框架。

（1）Deploy层：该层主要涉及了Flink的部署模式，Flink支持多种部署模式：本地、集群（Standalone/YARN）、云（GCE/EC2），Standalone 部署模式与Spark类似。

（2）Runtime层：Runtime层提供了支持Flink计算的全部核心实现，比如：支持分布式Stream处理、Job Graph到Execution Graph的映射、调度等，为上层API层提供基础服务。

（3）API层：API层主要实现了面向无界Stream的流处理和面向Batch的批处理API，其中面向流处理对应DataStream API，面向批处理对应DataSet API。

（4）Libraries层：该层也可以称为Flink应用框架层，根据API层的划分，在API层之上构建的满足特定应用的实时计算框架，也分别对应于面向流处理和面向批处理两类。面向流处理支持：CEP（复杂事件处理）、SQL-like的操作（基于Table的关系操作）；面向批处理支持：FlinkML（机器学习库）、Gelly（图处理）。

图2.1 Flink的设计架构

Flink 的核心计算架构是图2.1中的 Flink Runtime 执行引擎，它是一个分布式系统，能够接受数据流程序并在一台或多台机器上以容错方式执行。Flink Runtime 执行引擎可以作为 YARN（Yet Another Resource Negotiator）的应用程序在集群上运行，也可以在 Mesos 集群上运行，还可以在单机上运行（这对于调试 Flink 应用程序来说非常有用）。

Flink 分别提供了面向流式处理的接口（DataStream API）和面向批处理的接口（DataSet API）。因此，Flink 既可以完成流处理，也可以完成批处理。Flink 支持的拓展库涉及机器学习（FlinkML）、复杂事件处理（CEP）、以及图计算（Gelly），还有分别针对流处理和批处理的 Table API。

能被 Flink Runtime 执行引擎接受的程序很强大，但是这样的程序有着冗长的代码，编写起来也很费力，基于这个原因，Flink 提供了封装在 Runtime 执行引擎之上的 API，以帮助用户方便地生成流式计算程序。Flink 提供了用于流处理的 DataStream API 和用于批处理的 DataSet API。值得注意的是，尽管 Flink Runtime 执行引擎是基于流处理的，但是 DataSet API 先于 DataStream API 被开发出来，这是因为工业界对无限流处理的需求在 Flink 诞生之初并不大。

DataStream API 可以流畅地分析无限数据流，并且可以用 Java 或者 Scala 等来实现。开发人员需要基于一个叫 DataStream 的数据结构来开发，这个数据结构用于表示永不停止的分布式数据流。

Flink 的分布式特点体现在它能够在成百上千台机器上运行，它将大型的计算任务分成许多小的部分，每个机器执行一部分。Flink 能够自动地确保发生机器故障或者其他错误时计算能够持续进行，或者在修复 bug 或进行版本升级后有计划地再执行一次。这种能力使得开发人员不需要担心运行失败。Flink 本质上使用容错性数据流，这使得开发人员可以分析持续生成且永远不结束的数据（即流处理）。

2.2 Flink的运行架构

图2.2 Flink的运行架构

Flink对数据的处理被抽象为以下三步：第一，接受数据；第二，处理数据；第三，输出处理结果。具体来说就是：接收（ingest）一个或者多个数据源（hdfs，kafka等）；执行若干用户需要的转换算子（transformation operators）；将转换后的结果输出（sink）。

如图2.2所示，Flink处理数据流的算子（operator）分为三类：Source负责管理输入（数据源）、Tranformation负责数据运算、Sink负责管理结果输出。

（1）Source：数据源，Flink 在流处理和批处理上的 source 大概有 4 类：基于本地集合的 source、基于文件的 source、基于网络套接字的 source、自定义的 source。自定义的 source 常见的有 Apache kafka、RabbitMQ 等，当然也可以定义自己的 source。
（2）Transformation：数据转换的各种操作，有 Map / FlatMap / Filter / KeyBy / Reduce / Fold / Aggregations / Window / WindowAll / Union / Window join / Split / Select等，操作很多，可以将数据转换计算成你想要的数据。
（3）Sink：接收器，Flink 将转换计算后的数据发送的地点 ，你可能需要存储下来，Flink 常见的 Sink 大概有如下几类：写入文件、打印出来、写入 socket 、自定义的 sink 。自定义的 sink 常见的有 Apache kafka、RabbitMQ、MySQL、ElasticSearch、Apache Cassandra、Hadoop FileSystem 等，同理你也可以定义自己的 sink。

2.3 Flink的系统架构

图2.3 Flink的系统架构

Flink的系统架构如图2.3所示。用户在客户端提交作业（Job）到服务端，服务端为分布式的主从架构。Master上的Dispatcher服务负责提供REST接口来接收Client提交的Job，运行Web UI，并负责启动和派发Job给JobManager。Resource Manager负责计算资源（TaskManager）的管理。JobManager负责将任务调度到TaskManager执行、检查点（checkpoint）的创建等工作，而TaskManager(worker)负责SubTask的实际执行。当服务端的JobManager接收到一个Job后，会按照各个算子的并发度将Job拆分成多个SubTask，并分配到TaskManager的Slot上执行。具体的任务执行流程如图2.4所示。

图2.4 Flink的任务提交流程示意图

重要概念：

Job：一个Job代表一个可以独立提交给Flink执行的作业，我们向JobManager提交任务的时候就是以Job为单位的，只不过一份代码里可以包含多个Job（每个Job对应一个类的main函数）。

图2.5 Task和SubTask的区别

图2.5中每个圆代表一个Operator（算子），每个虚线圆角框代表一个Task，每个虚线直角框代表一个Subtask，其中的p表示算子的并行度。
（1）最上面是StreamGraph，在没有经过任何优化时，可以看到包含4个Operator/Task：Task A1、Task A2、Task B、Task C。
（2）StreamGraph经过链式优化（Flink默认会将一些并行度相同的算子连成一条链）之后，Task A1和Task A2两个Task合并成了一个新的Task A（可以认为合并产生了一个新的Operator），得到了中间的JobGraph。
（3）然后以并行度为2（需要2个Slot）执行的时候，Task A产生了2个Subtask，分别占用了Thread #1和Thread #2两个线程；Task B产生了2个Subtask，分别占用了Thread #3和Thread #4两个线程；Task C产生了1个Subtask，占用了Thread #5。

由此可以总结如下：

Task是逻辑概念，一个Operator就代表一个Task（多个Operator被chain之后产生的新Operator算一个Operator）；真正运行的时候，Task会按照并行度分成多个Subtask，Subtask是执行/调度的基本单元；每个Subtask需要一个线程（Thread）来执行。
前一小节讲了TaskManager才是真正干活的，启动的时候，它会将自己的资源以Slot的方式注册到master节点上的资源管理器（ResourceManager）。JobManager从ResourceManager处申请到Slot资源后将自己优化过后的SubTask调度到这些Slot上面去执行。在整个过程中SubTask是调度的基本单元，而Slot则是资源分配的基本单元。需要注意的是目前Slot只隔离内存，不隔离CPU。

为了更高效地使用资源，Flink默认允许同一个Job中不同Task的SubTask运行在同一个Slot中，这就是SlotSharing。但需要注意几个关键条件：
（1）必须是同一个Job。这个很好理解，slot是给Job分配的资源，目的就是隔离各个Job，如果跨Job共享，但隔离就失效了；
（2）必须是不同Task的Subtask。这样是为了更好的资源均衡和利用。一个计算流中（pipeline），每个Subtask的资源消耗肯定是不一样的，如果都均分slot，那必然有些资源利用率高，有些低。限制不同Task的Subtask共享可以尽量让资源占用高的和资源占用低的放一起，而不是把多个高的或多个低的放一起。比如一个计算流中，source和sink一般都是IO操作，特别是source，一般都是网络读，相比于中间的计算Operator，资源消耗并不大。
（3）默认是允许sharing的，也就是你也可以关闭这个特性。

下面我们依次来看看官方文档给出的两幅图：

图2.6 并行度小于slot数
图2.6中两个TaskManager节点共有6个slot，5个SubTask，其中sink的并行度为1，另外两个SubTask的并行度为2。此时由于Subtask少于Slot个数，所以每个Subtask独占一个Slot，没有SlotSharing。

图2.7 并行度大于slot数
当把并行度改为6，如图2.7所示，此时，Subtask的个数多于Slot了，所以出现了SlotSharing。一个Slot中分配了多个Subtask，特别是最左边的Slot中跑了一个完整的Pipeline。SlotSharing除了提高了资源利用率，还简化了并行度和Slot之间的关系：一个Job运行需要的最少的Slot个数就是其中并行度最高的那个Task的并行度，并行度最高和作业的最大并行度没有任何关系。