1.local（本地）——单机模式
2.standalone ——独立模式，Flink 自带集群，开发测试环境使用
3.yarn——计算资源统一由 Hadoop YARN 管理，生产环境测试
[warningbox title="前置需求"]
1. jdk1.8 及以上 配置JAVA_HOME 环境变量
2. ssh 免密码登录 集群内节点之间免密登录
[/warningbox]
| JobManager(master) | TaskManager(worker) | |
| node01 | * | * |
| node02 | * | |
| node03 | * |
下载地址:https://archive.apache.org/dist/flink/
此次使用flink-1.7.2-bin-hadoop26-scala_2.11.tgz
1.解压 Flink 压缩包到指定目录
2.修改安装目录下 conf 文件夹内的 flink-conf.yaml 配置文件，
# 指定 JobManager# 配置 Master 的机器名（ IP 地址）
jobmanager.rpc.address: node01
# 配置每个 taskmanager 生成的临时文件夹
taskmanager.tmp.dirs: /export/servers/flink-1.7.2/tmp3.配置 Slaves 节点，修改安装目录下 conf 文件夹内的 slave 配置文件， 指定 TaskManager
node01
node02
node03
4.修改各个节点环境变量文件/etc/profile，添加
export HADOOP_CONF_DIR=/export/servers/hadoop-2.6.0-cdh5.14.0/etc/hadoop
source /etc/profile4.分发 Flink 到各个节点
for i in {2..3}; do scp -r flink-1.7.2/ node0$i:$PWD; done5.启动集群
bin/start-cluster.sh
web-ui端口:8081
jps
--------------------- node01 ----------------
86583 Jps
85963 StandaloneSessionClusterEntrypoint
86446 TaskManagerRunner
--------------------- node02 ----------------
44099 Jps
43819 TaskManagerRunner
--------------------- node03 ----------------
29461 TaskManagerRunner
29678 Jps
[successbox title="基础配置"]
# jobManager 的 IP 地址
jobmanager.rpc.address: localhost
# JobManager 的端⼝号
jobmanager.rpc.
port: 6123
# JobManager JVM heap 内存⼤⼩
jobmanager.heap
.size: 1024m
# TaskManager JVM heap 内存⼤⼩
taskmanager.heap
.size: 1024m
# 每个 TaskManager 提供的任务 slots 数量⼤⼩
taskmanager.numberOfTaskSlots: 1
# 程序默认并⾏计算的个数
parallelism.default: 1[/successbox]
启动/停止 flink 集群
启动： ./bin/start-cluster.sh
停止： ./bin/stop-cluster.shFlink 集群的重启或扩容
启动/停止 jobmanager
如果集群中的 jobmanager 进程挂了， 执行下面命令启动
bin/jobmanager.sh start
bin/jobmanager.sh stop
启动/停止 taskmanager 添加新的 taskmanager 节点或者重启 taskmanager 节点
bin/taskmanager.sh start
bin/taskmanager.sh stopStandalone 集群架构
client 客户端提交任务给 JobManager
JobManager 负责 Flink 集群计算资源管理，并分发任务给 TaskManager
执行TaskManager 定期向 JobManager 汇报状态
| JobManager(master) | TaskManager(worker) | |
| node01 | * | * |
| node02 | * | * |
| node03 | * |
1.在 flink-conf.yaml 中添加 zookeeper 配置
#开启 HA， 使用文件系统作为快照存储
state.backend: filesystem
#默认为 none， 用于指定 checkpoint 的 data files 和 meta data 存储的目录
state.checkpoints.dir: hdfs://node01:8020/flink-checkpoints
#默认为 none， 用于指定 savepoints 的默认目录
state.savepoints.dir: hdfs://node01:8020/flink-checkpoints
#使用 zookeeper 搭建高可用
high-availability: zookeeper
# 存储 JobManager 的元数据到 HDFS,用来恢复 JobManager 所需的所有元数据
high-availability.storageDir: hdfs://node01:8020/flink/ha/
high-availability.zookeeper.quorum: node01:2181,node02:2181,node03:2181
2.将配置过的 HA 的 flink-conf.yaml 分发到另外两个节点并对node02的flink-conf.yaml进行修改
for i in {2..3}; do scp -r /export/servers/flink-1.7.2/conf/flink-conf.yaml node0$i:$PWD; donejobmanager.rpc.address: node023.在 masters 配置文件中添加多个节点并进行分发
node01:8081
node02:8081for i in {2..3}; do scp -r /export/servers/flink-1.7.2/conf/masters node0$i:$PWD; done4.启动
#各节点zookeeper启动 HDFS 集群
zkServer.sh start
bin/start-cluster.sh
[warningbox title="前置需求"]
1.jdk1.8 及以上【配置 JAVA_HOME 环境变量】
2).ssh 免密码登录【集群内节点之间免密登录】
3.至少 hadoop2.2 hdfs & yarn
[/warningbox]
yarn-session 提供两种模式
1.会话模式 使用 Flink 中 的 yarn-session （ yarn 客 户 端 ） ， 会 启 动 两 个 必 要 服 务 JobManager 和 TaskManagers 客户端通过 yarn-session 提交作业 yarn-session 会一直启动，不停地接 收客户端提交的作用 ，有大量的小作业，适合使用这种方式。
2.分离模式 直接提交任务给 YARN ,大作业，适合使用这种方式
[successbox title="yarn session 会话模式"]
yarn-session.sh(开辟资源)+flink run(提交任务) 这种模式下会启动 yarn session，
并且会启动 Flink 的两个必要服务： JobManager 和 Task-managers，然后你可以向集群提交作业。
同一个 Session 中可以提交多个 Flink 作业。需要注意的是，这种模式下 Hadoop 的版本至少 是 2.2，
而且必须安装了 HDFS（因为启动 YARN session 的时候会向 HDFS 上 提交相关的 jar 文件和配置文件）
通过./bin/yarn-session.sh 脚本启动 YARN Session 脚本可以携带的参数
-n,--container 分配多少个 yarn 容器 (=taskmanager 的数量)
Optional -D 动态属性
-d,--detached 独立运行 （以分离模式运行作业）
-id,--applicationId YARN 集群上的任务 id，附着到一个后台运行的 yarn session 中
-j,--jar Path to Flink jar file
-jm,--jobManagerMemory JobManager 的内存 [in MB]
-m,--jobmanager 指定需要连接的 jobmanager(主节点)地址 ,使用这个参数可以指定一 个不同于配置文件中的 jobmanager
-n,--container 分配多少个 yarn 容器 (=taskmanager 的数量)
-nm,--name 在 YARN 上为一个自定义的应用设置一个名字
-q,--query 显示 yarn 中可用的资源 (内存, cpu 核数)
-qu,--queue 指定 YARN 队列
-s,--slots 每个 TaskManager 使用的 slots 数量
-st,--streaming 在流模式下启动 Flink
-tm,--taskManagerMemory 每个 TaskManager 的内存 [in MB]
-z,--zookeeperNamespace 针对 HA 模式在 zookeeper 上创建 NameSpace 注:如果不想让 Flink YARN 客户端始终运行，那么也可以启动分离的 YARN 会话。 该参数被称为 -d 或--detached
启动： bin/yarn-session.sh -n 2 -tm 800 -s 1 -d
上面的命令的意思是，同时向 Yarn 申请 3 个 container（即便只申请了两个，因为 ApplicationMaster 和 Job Manager 有一个额外的容器。一旦将 Flink 部署到 YARN 群集 中，它就 会显示 Job Manager 的连接详细信息），其中 2 个 Container 启动 TaskManager （-n 2），每个 TaskManager 拥有 1 个 Task Slot（-s 1），并且向每个 TaskManager 的 Container 申请 800M 的内存，以及一个 ApplicationMaster（Job Manager）。 启动成功之后，去 yarn 页面：ip:8088 可以查看当前提交的 flink session
停止当前任务： yarn application -kill application_1527077715040_0007
[/successbox]
[successbox title="在YARN上运行一个Flink作业"]
上面的 YARN session 是在 Hadoop YARN 环境下启动一个 Flink cluster 集群，里面的资源 是可以共享给其他的 Flink 作业。我们还可以在 YARN 上启 动一个 Flink 作业，这里我们还是使 用./bin/flink，但是不需要事先启动 YARN session.
使用 flink 直接提交任务
bin/flink run -m yarn-cluster -yn 2 ./examples/batch/WordCount.jar
以上命令在参数前加上 y 前缀，-yn 表示TaskManager个数
停止 yarn-cluster yarn application -kill application 的 ID
注意：如果使用的 是 flink on yarn 方式，想切换回 standalone 模式的话， 需要删除文件： 【/tmp/.yarn-properties-root】 因为默认查找当前 yarn 集群中已有的 yarn-session 信息中的 jobmanager
运行方式：flink run -m yarn-cluster
[/successbox]
文章评论