Flink系统组成

Flink是一个分层系统，从下到上分为：系统部署层、任务运行层、API层以及基于API开发的通用库层(Libraries)。

flink组成

Flink部署层

Flink部署模式分为：单机部署、集群部署和Cloud部署。我们可以在单机上来启动Flink服务作为测试学习使用；也可以将Flink与通用资源管理器YARN、Mesos、K8s以及自身的提供资源管理器的Standalone模式来部署Flink集群；同时Flink支持部署在EC2、GCE等云服务上。

任务运行层

Flink的系统核心就是任务的运行引擎层，运行层接收执行任务JobGraph，JobGraph类似Storm中的Topology(Flink中有好几种Graph概念，之后会说)。我们下面所说的系统架构，就是Flink的任务运行层。

API层

Flink同时支持流式作业和批处理作业，DataStream是用于编写流式任务的API，DataSet是用于编写批处理作业的API。DataSet API和DataStream API通过各自的编译器会都会生成JobGraph，DataSet API使用优化器来生成最佳的执行计划JobGraph，DataStream API使用流构建器来生成JobGraph。

通用库层

基于DataStream API和DataSet API，Flink提供了一些通用的应用库，比如基于DataStream API开发的用于事件处理的CEP、用于逻辑表查询的Table API和SQL，基于DataSet API开发的机器学习库FlinkML、用于图计算的Gelly、用于逻辑表查询的Table API和SQL(Table API和SQL即支持批处理作业，也支持流处理作业)。

Flink系统架构

我们所说的Flink系统架构，其实对应的就是Runtime层。Flink作为分布式系统，它的系统架构和其它分布式大数据系统架构基本一致。采用主从架构，Master节点启动的服务称为JobManager(作业管理器)，Worker节点启动的服务称为TaskManager(任务管理器)。

Flink Runtime架构

Flink集群节点之间的服务是通过Akka actor进行通信的，每个服务节点都会启动一个ActorSystem来管理本节点的Actor。

JobManager

JobManager作为Flink集群Master节点，集群中至少需要启动一个JobManager，如果配置HA则可能会启动多个JobManager，但是在同一时刻只会存在一个leader，其它JobManager作为Standby，如果leader挂掉后会从其它Standby JobManager重新选举一个leader JobManager。

JobManager主要负责任务调度(sechedule task)、管理检查点(coordinate checkpoint)和故障恢复(recovery on failure)等任务。如果Flink采用Standalone模式部署集群，JobManager还会负责资源管理(resource manager)。

TaskManager

TaskManager作为Flink集群的Worker节点，集群中至少需要启动一个TaskManager。TaskManager的主要作用就是接收JobManager发送的task，并且向JobManager反馈任务执行信息。

TaskManager负责执行Task，Task是DataFlow任务的一个个子任务(subtask，算子的一个并行度)。这些Task在同一个JVM(TaskManager进程)中执行，这样做的好处是所有Task可以共用TCP连接、共享节点间的心跳信息以及减少网络传输。TaskManager还会负责缓冲(buffer)和交换(exchange)数据流。这里需要注意的是TaskManager之间的数据传输，使用的是Netty而不是Akka。

TaskManager并不是执行任务的最细粒度，为了控制每个TaskManager能够接收多少个task，Flink提供了Task Slot概念。

Task Slot

TaskManager中执行task的容器是Task Slot。每个Task Slot都是一个固定大小资源的子集。比如我们TaskManager启动了三个Task Slot，则每个Task Slot会获得TaskManager所管理内存的1/3。这样来自不同job的task不会为了内存而竞争，从而完成了内存资源的隔离。需要注意的是slot并没有对CPU进行隔离。

Task Slot

使用Slot的优点：

• Flink集群中的Slot个数与job中最高的并行度一直，从而简化了性能调优的过程。
• 允许多个Task共享Slot，提升了资源利用率。比如我们读取Kafka的topic有2个partition，那么对应了2个task，而map设置了并行度20，如果不可以共享Slot，就需要22个slot了。
• 可以起到资源隔离的目的，防止多个Job的task竞争内存。

共享Slot

共享Slot虽然能够提供资源利用率，但是如果一个Slot容纳太多的task反而会影响执行效率。默认情况共享Slot是开启的，共享Slot的task可以来自不同的task子集，比如下图中source task1和map task1共享一个task slot。共享slot会保证所共享的task来自同一个作业(job)。

共享slot

Source和map是一个Operator Chain，关于Operator Chain之后会说到

为了防止同一个slot包含太多的task，Flink提供了资源组(group)的概念。group就是对operator进行分组，同一group的不同operator task可以共享同一个slot。默认所有operator属于同一个组"default"，也就是所有operator task可以共享一个slot。我们可以通过slotSharingGroup()为不同的operator设置不同的group。

dataStream.filter(...).slotSharingGroup("groupName");

Slot 与任务并行度

下图是官方给出的slot配置方式和slot与任务并行度(parallelism)的关系。

TaskManager中的slot个数在flink-conf.yaml中配置，配置项为taskmanager.numberOfTaskSlots。建议slot个数最好与TaskManager的CPU数相同。

任务并行度的最大值由TaskManager集群的Slot总数决定，如果集群中slot总数为9，那么任务中算子组件最大的并行度也只能为9。
任务并行度如果没有显示给出，则默认所有算子的并行度都为1。我们可以通过以下三种方式指定任务的并行度：

1. 配置flink-conf.yaml中的parallelism.default，默认为1。
2. 使用flink命令行指定并行度bin/flink -p 2。
3. 在程序中指定，可以通过StreamExecutionEnvironment.setParallesim()为所有算子指定并行度，也可以为特定算子指定并行度map(..).setParallesim()。

slot与并行度

Client

Client并不属于Runtime架构中的一部分，但是它是用户提交作业到JobManager的桥梁。Client提交作业后可以与Flink集群断开，也可以一直连接来获取执行进度报告。
Client主要负责接收用户编写的Flink作业、解析作业执行计划、优化执行计划，然后提交作业给JobManager。

Client解析作业执行计划的过程就是将每一类算子解析成operator，然后根据算子之间的关系将所有operator组合起来，形成一个StreamGraph。
Client解析完执行计划后，接下来还会对执行计划进行优化。而优化的方式就是生成Operator Chain(关于Operator Chain之后会介绍到)，优化之后的执行计划称为JobGraph。
最后Client通过Akka actor将JobGraph提交给JobManager。