Flink Standalone集群部署和任务提交

Flink Standalone集群部署和任务提交

Flink按部署模式（也就是application运行模式）分为Flink Application Cluster（in Application Mode）、Flink Job Cluster（in a Per-Job Mode）、Flink Session Cluster（in Session Mode）三种模式。而按资源的提供者（Resource Providers）可以分为Standalone、Native Kubernetes、Yarn。

这里我们搭建的Standalone就是Flink Session Cluster的模式。在Standalone Cluster下，Per-Job Mode模式是不支持的，Application Mode模式也必须用standalone-job.sh来提交，用 ./flink run-application也是不支持的，可以用 ./flink run来提交。

部署Standalone Cluster之前的环境准备：

jdk1.8及以上

flink安装包（注意scala的版本）

安装部署步骤

1. 将安装包解压，拷贝到具体的安装位置（比如 /opt/install/flink-1.12.7/），修改配置，这里只涉及关键配置，其他配置可自行官网查看：

https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/deployment/config/

假设机器有hadoop1、hadoop2、hadoop3、hadoop4

2. 修改conf/masters和conf/workers文件

conf/masters是jobmanager的机器列表，格式：hostname:port

这里配置为

hadoop1:8081

conf/workers是taskmanager的机器列表，格式:hostname

这里配置为

hadoop2hadoop3hadoop4

3. 下面配置conf/flink-conf.yaml文件

jobmanager.rpc.address 是jobmanager rpc通信绑定的地址，这里配置为jobmanager的主机名，即hadoop1

jobmanager.rpc.port 是jobmanager的rpc端口，默认是6123

jobmanager.memory.process.size 是 jobmanager jvm进程的堆内存大小，默认是1600m

taskmanager.memory.process.size 是taskmanager jvm进程的堆内存大小，默认是1728m

taskmanager.numberOfTaskSlots 表示每个taskmanager所能提供的slots数量，也就是flink节点的计算能力，默认值1

以上关于资源的设置要根据集群大小以及机器配置来按需设置。

parallelism.default 是默认任务的并行度，任务没指定时即用该默认值

经过以上基本配置，然后将配置好的安装包拷贝到其他节点，这里是hadoop2-4，然后就可以启动Standalone集群。

4. 启动集群：

./bin/start-cluster.sh

（此处注意不要用 sh bin/start-cluster.sh 启动会报错，原因请看sh和./ 或bash的区别）

start-cluster.sh里面调用的就是jobmanager.sh和taskmanager.sh

另外jobmanager机器即hadoop1要和hadoop2-4打通ssh免密登录，不然无法ssh到hadoop2-4上taskmanager就无法正常启动。没打通ssh免密，则需要手动启动taskmanager。

单独启动或关闭 taskmanager，启动后对应的进程名是TaskManagerRunner

./bin/taskmanager.sh start/stop

单独启动或关闭 jobmanager，启动后对应的进程名是StandaloneSessionClusterEntrypoint

./bin/jobmanager.sh start/stop

启动后可以访问浏览器界面查看web ui，这里是：http://hadoop1:8081

5. 停止集群：

./bin/stop-cluster.sh

提交任务

以Session Mode模式提交

./bin/flink run xxx

以Application Mode模式提交（Flink Application Cluster和Application的生命周期一致，所以该方式提交和上面的Session Cluster都没关系）

./bin/standalone-job.sh start --job-classname xxx

配置Standalone Cluster的HA

配置ha，需要zookeeper、hdfs（分布式或共享的文件系统）的的支持

1. 修改flink-conf.yaml文件，

high-availability: zookeeperhigh-availability.zookeeper.quorum: hadoop1:2181,hadoop2:2181,hadoop3:2181high-availability.zookeeper.path.root: /flinkhigh-availability.cluster-id: /cluster_one # important: customize per clusterhigh-availability.storageDir: hdfs://ns/flink/recovery

2. 修改masters文件，即写多个，这里假设是hadoop1和hadoop0

hadoop1:8081hadoop2:8081

3. 启动集群

./bin/start-cluster.sh