文章目录

Flink部署

Standalone模式

flink-conf.yaml

masters

slaves

bin目录

提交作业

parallelism优先级设置

Yarn模式

Session-cluster

Per-Job-Cluster

其它

Flink架构

flink运行时组件

任务提交流程

任务调度原理

并行度(Parallelism)

Task&Slot

程序与数据流

执行图

并行度

任务链(Operator Chains)

Flink部署

本处主要记录standalone和on yarn两种部署方式

Standalone模式

conf 目录下调整文件

主要关注如下文件

flink-conf.yaml

通用配置

jobmanager地址

jobmanager.rpc.address: localhost

jobmanager端口号

jobmanager.rpc.port: 6123

jobmanager jvm 堆内存

jobmanager.heap.size: 1024m

TaskManager堆内存

taskmanager.heap.size: 1024m

一个taskmanager的槽位数

taskmanager.numberOfTaskSlots: 3

默认的并行度

parallelism.default: 1

故障恢复策略

jobmanager.execution.failover-strategy: region

当拥有多个并行度和任务，某一个分支挂了后只重启这个分支影响的任务，不用重启整个大任务

具体参数请参考：https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/ops/config.html#collapse-138

masters

localhost:8081

8081位网页访问的端口号，localhost为主机名

slaves

localhost

localhost为taskmanager的地址，可以设置多个

bin目录

启动命令

.bin/start-cluster.sh

启动后会出现如下信息

Starting cluster.

Starting standalonesession daemon on host MacBook.local.

Starting taskexecutor daemon on host MacBook.local.

其中 standalonesession daemon 为 Jobmanager；taskexecutor daemon 为task manager

启动完成后 执行 jps 命令

28330 StandaloneSessionClusterEntrypoint

28764 TaskManagerRunner

StandaloneSessionClusterEntrypoint 为jobmanager相关的提交job入口点

TaskManagerRunner为启动taskmanager的runner

启动完成后 进入 http://localhost:8081/#/overview 页面可以访问前端页面

停止命令

.bin/stop-cluster.sh

参考文档

提交作业

提交作业分页面提交和命令提交

页面提交

http://localhost:8081/#/submit 页面提交

命令提交

提交 ./bin/flink run -c com.ffl.study.flink.scala.stream.StreamWordCount -p 2 /Users/feifeiliu/Documents/work/java/study/study-flink/target/study-flink-1.0-SNAPSHOT.jar --host localhost --port 7777

-c 为运行的类入口；-p为并行度;然后紧跟 jar文件路径 和 参数

查看所有job

./bin/flink list 查看运行任务

./bin/flink list -a 查看所有任务

停止 ./bin/flink cancel f1f07f2b6f308633479d85141b1aef80

parallelism优先级设置

代码中每个算子并行度

env.setParallelism() 设置的并行度

提交代码时设置的并行度

全局设置的并行度，即parallelism.default参数

参考文档

Yarn模式

启动需要在lib目录下加入 flink-shaded-hadoop-2-uber-2.8.3-10.0.jar 包，2.8.3 为具体hadoop版本号，可以自己选择

Session-cluster

Session-cluster 在yarn中初始化一个flink集群，开辟指定的资源，以后提交任务都向这里提交。这个flink集群会常驻在yarn集群中，除非手动停止

启动命令

启动hadoop集群，略

启动yarn-session

bin/yarn-session.sh -n 3 -s 3 -jm 1024 -tm 1024 -nm bjsxt -d

-n 为 taskManager数量，以后版本可以忽略，动态分配

-s 每个taskManager里slot数量

-jm,–jobManagerMemory :设置 JobManager 的内存，单位是 MB。

-tm,–taskManagerMemory :每个 TaskManager 的内存，单位是 MB。

-nm 在 YARN 上为一个自定义的应用设置一个名字

-d 后台执行

3. 提交作业

./bin/flink run -c com.ffl.study.flink.scala.stream.StreamWordCount -p 2 /Users/feifeiliu/Documents/work/java/study/study-flink/target/study-flink-1.0-SNAPSHOT.jar --host localhost --port 7777

取消yarn-session

yarn application --kill xxxx

Per-Job-Cluster

每次提交都会创建一个新的flink集群，任务之间互相独立，互不影响，方便管理。任务执行完成后创建的集群也会消失

启动hadoop集群，略

不启动yarn-session，直接执行job

./bin/flink run -m yarn-cluster -c com.ffl.study.flink.scala.stream.StreamWordCount -p 2 /Users/feifeiliu/Documents/work/java/study/study-flink/target/study-flink-1.0-SNAPSHOT.jar --host localhost --port 7777

参考文档

其它

kubernetes

参考文档

Docker

参考文档

Flink架构

flink运行时组件

JobManager 作业管理器

控制一个应用程序执行的主进程，也就是说，每个应用程序都会被一个不同的JobManager所控制执行

Jobmanager会先接受到要执行的执行程序，这个执行程序包括：作业图(JobGraph)、逻辑流程图(logical dataflow graph)和打包了的所有类库和其它资源的jar包

JobManager会把JobGraph转换成一个物理层面的数据流图，这个图被叫做“执行图”(ExecutionGraph)，包含了所有可以并发执行的任务

JobManager会向资源管理器(ResourceManager)请求执行任务必要的资源，也就是任务管理器(TaskManger)上的插槽(slot)。一旦获取到足够资源，就会将执行图分发到真正运行它们的TaskManger上。而在运行过程中，JobManager会负责所有需要中央协调的操作，比如说检查点(checkpoint)的协调

TaskManager 任务管理器

Flink中工作进程。通常在Flink中会有多个TaskManager运行，每一个TaskManager都包含一定数量的插槽(slots)。插槽的数量限制了TaskManager能够执行的任务数量

启动之后，TaskManager会向资源管理器注册它的插槽；收到资源管理器的指令后，TaskManager就会将一个或多个插槽提供给JobManager调用。JobManager就可以向插槽分配任务(task)来执行了。

在执行过程中，一个TaskManager可以跟其它运行同一个程序的TaskManager交换数据

ResourceManager 资源管理器

主要负责任务管理器的插槽(slot),TaskManager插槽是Flink中定义的处理资源单元

Dispacher 分发器

可以跨作业运行，它为应用提交提供了REST接口

在架构中不是必须的，这取决于应用提交流程的方式

任务提交流程

分普通提交流程和on yarn模式

任务提交流程一

任务提交流程(Yarn)

参考文档

任务调度原理

参考文档

思考

怎样实现并行计算？

并行的任务，需要占用多少slot？

一个流处理程序，到底包含多少个任务？

先了解如下概念再作解答

并行度(Parallelism)

一个特定算子的子任务(subTask)的个数被称之为其并行度(Parallelism)

一般情况下，一个stream的并行度，可以认为就是其所有算子中最大的并行度

参考文档

Flink中每一个TaskManager都是一个JVM进程，它可能会在独立的线程上执行一个或多个子任务

为了控制一个TaskManager能接收多个Task，TaskManager通过task slot来进行控制(一个TaskManager至少有一个slot)

Task&Slot

默认情况下，Flink允许子任务共享slot，及时它是不同任务的子任务，这样的结果是，一个slot可以保存整个作业的管道

Task Slot是静态的概念，是指TaskManager具有的并发执行的能力

参考文档

所有Flink程序都由三部分组成：Source、Transformation和Sink

程序与数据流

在运行程序时，Flink上运行的程序会被映射成“逻辑数据流”(dataflows)，它包含了这三部分

每个dataflow以一个或多个sources开始以一个或多个sink结束，dataflow类似于有向无环图(DAG)

在大部分情况下，程序中的转换运算(transformations)跟dataflow中算子(operator)是一一对应的关系

执行图

Flink中执行图可以分为四层：StreamGraph -> JobGraph -> ExecutionGraph ->物理执行图

StreamGraph:是根据用户通过Stream Api编写的代码生成的最初的图，用来表示程序的拓扑结构

JobGraph：StreamGraph经过优化后生成了JobGraph，提交给JobManager的数据结构，主要优化为将多个符合条件的节点Chain在一起作为一个节点。即为UI界面上看到的图，在client上生成

ExecutionGraph：JobManager根据JobGraph生成ExecutionGraph。ExecutionGraph是JobGraph的并行化版本，是调度层的最核心的数据结构 (parallim中的图) 。JobManager上生成

物理执行图：JobManager根据ExecutionGraph对Job进行调度后，在各个TaskManager上部署Task后形成的“图”，并不是一个具体的数据结构。Task上运行

并行度

数据传输形式

一个程序中，不同的算子可能具有不同的并行度

算子之间的的传输形式可以是one-to-one(forwarding)的模式也可以是redistributing的模式，具体是哪一种形式，取决于算子的种类

one-to-one：stream维护着分区以及元素的顺序(比如source和map之间)。这意味着map算子的子任务看到的元素的个数以及顺序跟source算子的子任务生产的元素的个数、顺序相同。map、filter、flatMap等算子都是one-to-to的对应关系

Redistributing：stream的分区会发生改变。每一个算子的子任务依据所选择的transformation发送数据到不同的目标任务。例如，keyBy基于hashCode重分区、而broadcast和rebalance会随机重新分区，这些算子都会引起redistribute过程，而redistribute过程就类似于Spark中的shuffle过程。如下两种方式会引起Redistributing

并行度不同，会进行轮询rebalance

前后算子为宽依赖(hash重分区)

任务链(Operator Chains)

Flink采用了一种成为任务链的优化技术，可以在特定条件下减少本地通信的开销。为了满足任务链的要求，必须将两个或多个算子设为相同的并行度，并通过本地转发(local forward)的方式进行连接

相同并行度的one-to-one操作，flink这样相连的算子链接在一起形成一个task，原来的算子成为里面的subtask

并行相同、并且是one-to-one操作，两者缺一不可

https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/internals/job_scheduling.html

https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/learn-flink/datastream_api.html