Flink1.13.6详细部署方式

        Flink1.13.6支持flink cdc2.x版本，为兼容flink cdc，本文选择使用flink1.13.6版本部署。其他版本也可借鉴。

Flink支持多种部署方式

• local（本地）->单机部署，一般不会使用

• standalone（独立部署）->flink自带的部署方式，一般用于开发测试使用

• yarn（分布式部署）->由hadoopyarn统一的管理资源，是生产环境采用的部署方式

本文将针对以上几种方式进行详细的介绍：

 1、伪分布部署

安装步骤：

（1）下载安装包

https://archive.apache.org/dist/flink/flink-1.13.6/flink-1.13.6-bin-scala_2.11.tgz

（2）将压缩包上传到node1服务器

（3）解压缩安装包

tar -zxvf flink-1.13.6-bin-scala_2.11.tgz -C /opt/bigdata

（4）进入安装目录，启动flink

 ./start-cluster.sh 

（5）验证是否启动成功

jps

 

 （6）打开监控页面

http://node1:8081

（7）运行测试任务

bin/flink run /opt/bigdata/flink-1.13.6/examples/batch/WordCount.jar

2、Standalone部署

步骤：

• 客户端向JobManager递交任务请求

• JobManager进行从节点的资源管理和Job任务的拆分，将拆分好的任务份发给从节点去执行

• 从节点定期的向JobManager汇报心跳信息和任务的执行状态

资源规划

• node1（Master——slave）

• node2（slave）

• node3（slave）

安装步骤:

（1）上传flink压缩包到指定目录

（2）解压缩flink到/opt/bigdata目录

tar -zxvf flink-1.13.6-bin-scala_2.11.tgz -C /opt/bigdata

（3）使用vi修改conf/flink-conf.yaml

# jobManager 的IP地址
jobmanager.rpc.address: node1

# JobManager 的端口号
jobmanager.rpc.port: 6123

# JobManager JVM heap 内存大小
jobmanager.heap.size: 1024

# TaskManager JVM heap 内存大小
taskmanager.heap.size: 1024

# 每个 TaskManager 提供的任务 slots 数量大小
taskmanager.numberOfTaskSlots: 2

#是否进行预分配内存，默认不进行预分配，这样在我们不使用flink集群时候不会占用集群资源
taskmanager.memory.preallocate: false

# 程序默认并行计算的个数
parallelism.default: 1

#JobManager的Web界面的端口（默认：8081）
jobmanager.web.port: 8081

#配置每个taskmanager生成的临时文件目录（选配）
taskmanager.tmp.dirs: /opt/bigdata/flink-1.13.6/tmp

slot和parallelism总结 :

taskmanager.numberOfTaskSlots:2

每一个taskmanager中的分配2个TaskSlot,3个taskmanager一共有6个TaskSlot

parallelism.default:1 运行程序默认的并行度为1，6个TaskSlot只用了1个，有5个空闲

slot是静态的概念，是指taskmanager具有的并发执行能力2。parallelism是动态的概念，是指程序运行时实际使用的并发能力

（4）使用vi修改slaves文件

node1
node2
node3

（5）使用vi修改/etc/profile系统环境变量配置文件，添加HADOOP_CONF_DIR目录

export HADOOP_CONF_DIR=/opt/bigdata/hadoop-2.6.0/etc/hadoop

（6）分发/etc/profile到其他两个节点

scp -r /etc/profile node2:/etc
scp -r /etc/profile node3:/etc

（7）每个节点重新加载环境变量

source /etc/profile

（8）使用scp命令分发flink到其他节点

scp -r /opt/bigdata/flink-1.13.6/ node2:/opt/bigdata/
scp -r /opt/bigdata/flink-1.13.6/ node3:/opt/bigdata/

（9）启动Flink集群

./bin/start-cluster.sh

启动/停止flink集群

• 启动：./bin/start-cluster.sh

• 停止：./bin/stop-cluster.sh

启动/停止jobmanager如果集群中的jobmanager进程挂了，执行下面命令启动

• bin/jobmanager.sh start

• bin/jobmanager.sh stop

启动/停止taskmanager添加新的taskmanager节点或者重启taskmanager节点

• bin/taskmanager.sh start

• bin/taskmanager.sh stop

 （10）在HDFS中创建/test/input目录

hdfs dfs -mkdir -p /test/input

（11）上传README.txt文件到HDFS /test/input目录

hdfs dfs -put /opt/bigdata/flink-1.13.6/README.txt /test/input

（12）运行测试任务

bin/flink run /opt/bigdata/flink-1.13.6/examples/batch/WordCount.jar \
--input hdfs://node1:8020/test/input/README.txt \
--output hdfs://node1:8020/test/output2/result.txt

（13）浏览Flink Web UI界面

http://node1:8081

3、standalone的ha环境部署

        从上述架构图中，可发现JobManager存在单点故障，一旦JobManager出现意外，整个集群无法工作。所以，为了确保集群的高可用，需要搭建Flink的HA。

 

资源规划

• node1（master+slave）

• node2（master+slave）

• node3（slave）

环境要求

• 按照zk集群，同时启动zk集群

• 安装hadoop集群，同时启动hadoop集群

安装步骤

（1）在flink-conf.yaml中添加zookeeper配置

#开启HA，使用文件系统作为快照存储
state.backend: filesystem
#启用检查点，可以将快照保存到HDFS
state.backend.fs.checkpointdir: hdfs://node1:8020/flink-checkpoints
#使用zookeeper搭建高可用
high-availability: zookeeper
# 存储JobManager的元数据到HDFS
high-availability.storageDir: hdfs://node1:8020/flink/ha/
high-availability.zookeeper.quorum: node1:2181,node2:2181,node3:2181

（2）将配置过的HA的flink-conf.yaml分发到另外两个节点

scp -r /opt/bigdata/flink-1.13.6/conf/flink-conf.yaml node2:/opt/bigdata/flink-1.13.6/conf/
scp -r /opt/bigdata/flink-1.13.6/conf/flink-conf.yaml node3:/opt/bigdata/flink-1.13.6/conf/

（3）到 node2 中修改flink-conf.yaml中的配置，将JobManager设置为自己节点的名称

jobmanager.rpc.address: node2

（4）在node1的masters配置文件中添加多个节点

node1:8081
node2:8081

（5）分发masters配置文件到另外两个节点

scp -r /opt/bigdata/flink-1.13.6/conf/masters node2:/opt/bigdata/flink-1.13.6/conf/
scp -r /opt/bigdata/flink-1.13.6/conf/masters node3:/opt/bigdata/flink-1.13.6/conf/

（6）启动zookeeper集群

（7）启动HDFS集群

（8）启动flink集群

（9）分别查看两个节点的Flink Web UI

（10）kill掉一个节点，查看另外的一个节点的Web UI

注意事项

切记搭建HA，需要将第二个节点的jobmanager.rpc.address修改为node2

Standalone集群 使用场景：主要是开发测试阶段

4、yarn的部署

        在一个企业中，为了最大化的利用集群资源，一般都会在一个集群中同时运行多种类型的 Workload。因此 Flink 也支持在Yarn上面运行；flink on yarn的前提是：hdfs、yarn均启动

集群规划

JobManager: node1

WorkManager: node1 node2 node3

前提条件：

需要关闭yarn的虚拟内存检测，如果不关闭，假如flink任务在执行过程中，所使用的内存超过了yarn分配的虚拟内存大小，会直接将flink任务杀死，导致任务执行失败，因此需要关闭yarn的虚拟内存检测

操作步骤

（1）修改Hadoop的yarn-site.xml，添加该配置表示内存超过分配值，是否将任务杀掉。默认为true。运行Flink程序，很容易超过分配的内存。

<property>
	<name>yarn.nodemanager.vmem-check-enabled</name>
    <value>false</value>
</property>

（2）分发yarn-site.xml到其它服务器节点

scp yarn-site.xml node2:$PWD
scp yarn-site.xml node3:$PWD

（3）启动HDFS、YARN集群

start-all.sh

Flink On yarn的运行原理

 Flink on yarn提供了3种模式

4.1会话模式（yarn-seeion）

 使用步骤:

（1）在flink目录启动yarn-session

bin/yarn-session.sh -n 2 -tm 800 -s 1 -d

# -n 表示申请2个容器，
# -s 表示每个容器启动多少个slot
# -tm 表示每个TaskManager申请800M内存
# -d 表示以后台程序方式运行

可以通过 bin/yarn-session.sh --help 查看yarn-session.sh脚本可以携带的参数

   Required
        -n,--container <arg>               分配多少个yarn容器 (=taskmanager的数量)  
      Optional
        -D <arg>                        动态属性
        -d,--detached                    独立运行 （以分离模式运行作业）
        -id,--applicationId <arg>            YARN集群上的任务id，附着到一个后台运行的yarn session中
        -j,--jar <arg>                      Path to Flink jar file
        -jm,--jobManagerMemory <arg>     JobManager的内存 [in MB] 
        -m,--jobmanager <host:port>        指定需要连接的jobmanager(主节点)地址  
                                       使用这个参数可以指定一个不同于配置文件中的jobmanager  
        -n,--container <arg>               分配多少个yarn容器 (=taskmanager的数量) 
        -nm,--name <arg>                 在YARN上为一个自定义的应用设置一个名字
        -q,--query                        显示yarn中可用的资源 (内存, cpu核数) 
        -qu,--queue <arg>                 指定YARN队列
        -s,--slots <arg>                   每个TaskManager使用的slots数量
        -st,--streaming                   在流模式下启动Flink
        -tm,--taskManagerMemory <arg>    每个TaskManager的内存 [in MB] 
        -z,--zookeeperNamespace <arg>     针对HA模式在zookeeper上创建NameSpace

（2）访问yarn的webui, http://node1:8088/

（3） 使用flink提交任务

bin/flink run examples/batch/WordCount.jar

（4）如果程序运行完了，可以使用yarn application -kill application_id杀掉任务

yarn application -kill application_1573371647348_0002

这种方式递交作业，服务一直处于启动状态，任务运行完成以后，服务不会停止，所以节省了任务递交过程中启动服务的操作，从而提高了执行效率，所以在频繁的递交作业、小作业的的场景适合采用这种部署方式

 4.2、yarn-pre-job模式

使用步骤：

（1）使用flink直接提交任务

./bin/flink run -t yarn-per-job --detached ./examples/streaming/TopSpeedWindowing.jar

（2）查看Yarn的WEB UI, 发现任务结束, Yarn的状态是SUCCESSED

如果使用的 是 flink on yarn 方式，想切 换回 standalone 模式的话， 需要删除文件：

/tmp/.yarn-properties-root

因为默认查找当前 yarn 集群中已有的 yarn-session 信息中的 jobmanage

 4.3、yarn-application模式

        对于生产使用，建议以 Per-job 或 Application Mode 部署 Flink 应用程序，因为这些模式为应用程序提供了更好的隔离。

        Application Mode 将在 YARN 上启动 Flink 集群，其中应用程序 jar 的 main() 方法在 YARN 中的 JobManager 上执行。应用程序完成后，集群将立即关闭。您可以使用 yarn application -kill <ApplicationId> 或取消 Flink 作业来手动停止集群。

        与yarn-pre-job不一样的就是ApplicationMode需要将jar包上传到hdfs上执行，yarn会随机选择一台服务器启动flink的jobmanager容器，并执行main方法。

使用步骤：

（1）使用flink直接提交任务

./bin/flink run-application -t yarn-application hdfs:///streaming/TopSpeedWindowing.jar

 部署应用程序模式集群后，用户可以与其交互以执行取消或获取保存点等操作。

./bin/flink list -t yarn-application -Dyarn.application.id=application_XXXX_YY
# Cancel running job
./bin/flink cancel -t yarn-application -Dyarn.application.id=application_XXXX_YY <jobId>

注意，取消yarn-application集群上的作业将停止集群。