• 主从架构：hadoop(hdfs mapreduce yarn)、spark、hbase、flink
• 去中心化架构：zookeeper、Kafka

Filnk:

-主节点 JobManager:

1.接收客户端请求

2.管理TaskManager

3.向YARN申请资源

4.向TM分发需要执行的Task

-从节点:TaskManager

1.执行Task

2.向JM汇报状态

Client

1.Client向JM中的分发器提交任务,分发器还会启动一个WEBUI页面服务

2.分发器启动一个与提交任务对应的Jobmaster

3.Jobmaster向资源管理器申请任务需要的资源

4.当资源不够时注册新的TM(仅YARN模式),如果是Standalone模式资源不够时拒绝服务

5.新启动的TM向资源管理器注册自己的资源

6.资源管理器向TM发出提供资源的命令

7.TMTM向对应的JobMaster请求需要执行的任务

8.Jobmaster向TM分发任务并执行

大概流程:

1.向JM发送提交任务的请求,并提交具体任务

2.验证请求是否合法

3.向空闲TM分配任务

分发器(Dispatcher)

Dispatcher 主要负责提供一个 REST 接口，用来提交应用，并且负责为每一个新提交的作业启动一个新的 JobMaster 组件。Dispatcher 也会启动一个 Web UI，用来方便地展示和监控作业执行的信息。

作业处理器(JobMaster)

JobMaster 是 JobManager 中最核心的组件，负责处理单独的作业（Job），JobMaster和每个任务是一一对应的。

资源管理器(ResourceManager)

ResourceManager 主要负责资源的分配和管理，在 Flink 集群中只有一个。所谓“资源”，主要是指 TaskManager 的任务槽（task slots）。任务槽就是 Flink 集群中的资源调配单元，包含了机器用来执行计算的一组 CPU 和内存资源。每一个任务（Task）都需要分配到一个 slot 上执行。

任务管理器-TaskManager

Slot 任务执行槽

Task 并行执行 subtasks

我们把一个算子操作，“复制”多份到多个节点，数据来了之后就可以到其中任意一个执行。这样一来，一个算子任务就被拆分成了多个并行的“子任务”（subtasks），再将它们分发到不同节点，就真正实现了并行计算。在 Flink 执行过程中，每一个算子（operator）可以包含一个或多个子任务（operator subtask），这些子任务在不同的线程、不同的物理机或不同的容器中完全独立地执行

优化思路:移动计算不移动数据 离线情况下

尽量在数据本地进行数据计算 节省网络传输过程

Flink 代码不动数据流动

原因就在于流式数据本身是连续到来的、我们不会同时传输所有数据，这其实是更符合数据流本身特点的处理方式

代码中设置

我们在代码中，可以很简单地在算子后跟着调用 setParallelism()方法，来设置当前算子的并行度,这种方式设置的并行度，只针对当前算子有效：

stream.map(word -> Tuple2.of(word, 1L)).setParallelism(2);

我们也可以直接调用执行环境的 setParallelism()方法，全局设定并行度：

env.setParallelism(2);

这样代码中所有算子，默认的并行度就都为 2 了。我们一般不会在程序中设置全局并行度，因为如果在程序中对全局并行度进行硬编码，会导致无法动态扩容。

提交应用时设置

在使用 flink run 命令提交应用时，可以增加-p 参数来指定当前应用程序执行的并行度，它的作用类似于执行环境的全局设置：

bin/flink run –p 2 –c org.apache.flink.examples.java.wordcount.WordCount /export/server/flink/examples/batch/WordCount.jar

配置文件中设置

直接在集群的配置文件 flink-conf.yaml 中直接更改默认并行度：

parallelism.default: 2

优先级：算子 > 代码全局 > 命令行参数 > 配置文件

算子链（Operator Chain）

算子间的数据传输

• 一对一（One-to-one，forwarding）
• 重分区（Redistributing）
• 

在 Flink 中，并行度相同的一对一（one to one）算子操作，可以直接链接在一起形成一个“大”的任务（task），这样原来的算子就成为了真正任务里的一部分，每个 task会被一个线程执行。这样的技术被称为“算子链”（Operator Chain）。

将算子链接成 task 是非常有效的优化：可以减少线程之间的切换和基于缓存区的数据交换，在减少时延的同时提升吞吐量。

在代码中对算子做一些特定的设置：
// 禁用算子链
.map(word -> Tuple2.of(word, 1L)).disableChaining();
// 从当前算子开始新链
.map(word -> Tuple2.of(word, 1L)).startNewChain()

任务槽的共享

Flink 默认是允许 slot 共享的，如果希望某个算子对应的任务完全独占一个 slot，或者只有某一部分算子共享 slot，我们也可以通过设置“slot 共享组”（SlotSharingGroup）手动指定：

.map(word -> Tuple2.of(word, 1L)).slotSharingGroup(“1”);

任务槽和并行度的关系

同一个任务节点的并行子任务是不能共享 slot 的，所以允许 slot 共享之后，运行作业所需的 slot 数量正好就是作业中所有算子并行度的最大值。

同一个slot中不能有同一个任务的两个及以上的并行子任务

Flink On Yarn

• Yarn的资源可以按需使用，提高集群的资源利用率
• Yarn的任务有优先级，根据优先级运行作业
• 基于Yarn调度系统，能够自动化地处理各个角色的 Failover(容错)- JobManager 进程和 TaskManager 进程都由 Yarn NodeManager 监控- 如果 JobManager 进程异常退出，则 Yarn ResourceManager 会重新调度 JobManager 到其他机器- 如果 TaskManager 进程异常退出，JobManager 会收到消息并重新向 Yarn ResourceManager 申请资源，重新启动 TaskManager

运行机制

1)当启动一个Flink Yarn会话时，客户端首先会检查本次请求的资源是否足够。资源足够将会上传包含HDFS配置信息和Flink的jar包到HDFS。

2)随后客户端会向Yarn发起请求，启动applicationMaster,随后NodeManager将会加载有配置信息和jar包，一旦完成，ApplicationMaster(AM)便启动。

3)当JobManager and AM 成功启动时,他们都属于同一个container，从而AM就能检索到JobManager的地址。此时会生成新的Flink配置信息以便TaskManagers能够连接到JobManager。同时，AM也提供Flink的WEB接口。用户可并行执行多个Flink会话。

4)随后，AM将会开始为分发从HDFS中下载的jar以及配置文件的container给TaskMangers.完成后Fink就完全启动并等待接收提交的job。

Session模式 会话模式

频繁的提交小任务 会提交On yarn模式,

特点：需要事先申请资源，提前启动JobManager和TaskManager

优点：不需要每次提交作业都申请资源，而是使用已经申请好的资源，从而执行效率高

缺点：作业每次执行完不会释放资源，因此会一起占用资源

应用场景：比较适合于提交比较频繁的场景，小作业比较多

缺点:资源占用,Yarn会把资源分配给 spark或者hive,flink则会占用固定资源造成资源占用

Per-Job模式 分离模式

特点：每次提交作业都需要申请一次资源（每个作业都会创建一个JobManager）

优点：作业运行完成后，资源会被释放，

缺点：每次提交都重新申请资源，会影响执行效率

应用场景：适合作业比较少的场景，大作业的情况下

application模式

flink-1.11引入了一种新的部署模式，即 Application 模式。目前，flink-1.11 已经可以支持基于 Yarn 和 Kubernetes 的 Application 模式。

Session模式：所有作业共享集群资源，隔离性差，JM 负载瓶颈，main 方法在客户端执行。

Per-Job模式：每个作业单独启动集群，隔离性好，JM 负载均衡，main 方法在客户端执行。

通过以上两种模式的特点描述，可以看出，main方法都是在客户端执行，社区考虑到在客户端执行 main() 方法来获取 flink 运行时所需的依赖项，并生成 dataflow，提交到集群的操作都会在实时平台所在的机器上执行，那么将会给服务器造成很大的压力。尤其在大量用户共享客户端时，问题更加突出。

Application 模式把main方法任务提交到集群的JM上去运行,相较于Per-Job分离模式做了优化