一、flink部署模式

flink有三种部署模式
1.1 local本地模式
1.2 Sandalone独立集群模式
1.3 Flink on Yarn模式

二、flink部署

企业级生产环境选用Flink on Yarn模式部署

2.1 选择flink版本

使用flink1.13.5版本，官网下载包地址：https://flink.apache.org/downloads.html

2.2 安装部署

将下载的安装包上传至服务器目录：/usr/local/apps/
解压安装包：tar -zxvf flink-1.13.5.tar.gz
配置hadoop环境变量：vim etc/profile
Export HADOOP_CLASSPATH=’hadoop classpath’

配置原因见flink问题汇总，问题3
3、添加相关依赖包到lib下
有任务需要写入hive，添加相关依赖包：
https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/table/hive/overview/

https://mvnrepository.com/artifact/org.antlr/antlr-runtime/3.5.2

其中flink-sql-connector-hive-2.2.0_2.11-1.13-SNAPSHOT.jar是根据flink源码修改pom依赖后重新编译打包的jar包，原因见flink问题汇 问题6

4、修改flink-conf.yaml
根据需要添加相关配置信息，另外flink是根据”: ”来进行切割配置的，所以配置名称和值中间一定要用”: ”间隔，否则配置不生效。(一定要看清是有空格的)
jobmanager.rpc.address: localhost
jobmanager.rpc.port: 6123
jobmanager.memory.process.size: 1024m
taskmanager.memory.process.size: 1024m
taskmanager.numberOfTaskSlots: 2
parallelism.default: 1
env.java.home: /usr/java/jdk1.8.0_321
rest.address: 0.0.0.0
rest.bind-port: 8081-8096
rest.bind-address: 0.0.0.0
classloader.check-leaked-classloader: false
web.submit.enable: true
io.tmp.dirs: /data/flink/tmp
jobmanager.archive.fs.dir: hdfs://****-ha/completed-jobs/
state.checkpoints.dir: hdfs://****-ha/flink/flink-checkpoints
state.checkpoints.num-retained: 3
state.backend: filesystem
state.backend.fs.checkpointdir: hdfs://****-ha/flink/flink-checkpoints
state.backend.incremental: true
state.savepoints.dir: hdfs://****-ha/flink/flink-savepoints
restart-strategy.fixed-delay.attempts: 5
restart-strategy.fixed-delay.delay: 10 s
yarn.application-attempts: 5
yarn.reallocate-failed: true
yarn.maximum-failed-containers: 5
high-availability.zookeeper.quorum: ****-slave1:2181,****-slave2:2181,****-slave3:2181
high-availability.storageDir: hdfs://****-ha/flink/recovery
high-availability.zookeeper.path.root: /flink
high-availability: zookeeper
akka.framesize: 20m
env.java.opts: -Dlog4j2.formatMsgNoLookups=true -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:+UseFastAccessorMethods -XX:InitiatingHeapOccupancyPercent=45 -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/data02/applogs -XX:+PrintGCTimeStamps -Xloggc:/data02/applogs/gc.log -XX:+PrintGCDetails -XX:+PrintGCApplicationConcurrentTime -XX:+PrintGCApplicationStoppedTime -XX:+PrintHeapAtGC
classloader.resolve-order: parent-first

参数 io.tmp.dirs: /data/flink/tmp  需要在安装zookeeper的节点创建此临时目录，并赋予读写执行权限。
mkdir -p /data/flink/tmp
chmod 777 /data/flink/tmp

将flink1.13.5依赖包放入lib目录

libfb303-0.9.3.jar包位置：

三、flink on yarn

3.1 划分yarn队列专门为flink使用

scheduler.fair.user-as

yarn.scheduler.fair.allow-undeclared-pools

yarn.admin.acl

Yarn动态管理池配置前

Yarn动态管理池配置后

3.2 根据配置文件创建相关路径

根据flink-conf.yaml文件配置创建相关路径
mkdir -p /data/flink/tmp
sudo -u hdfs hdfs dfs -mkdir /flink
sudo -u hdfs hdfs dfs -mkdir /flink/flink-checkpoints
sudo -u hdfs hdfs dfs -mkdir /flink/flink-savepoints
sudo -u hdfs hdfs dfs -mkdir /flink/recovery
sudo -u hdfs hdfs dfs -mkdir /flink/flink-job

#给创建路径赋权限
sudo -u hdfs hadoop fs -chmod 777 /flink
sudo -u hdfs hadoop fs -chmod 777 /user
sudo -u hdfs hadoop fs -chmod 777 /flink/recovery
sudo -u hdfs hadoop fs -chmod 777 /flink/flink-checkpoints
sudo -u hdfs hadoop fs -chmod 777 /flink/flink-savepoints
sudo -u hdfs hadoop fs -chmod 777 /flink/flink-job

3.3 启动模式

Flink on Yarn运行任务模式有两种
1>会话模式 (Yarn Session Mode)：先在yarn上启动一个Flink集群，然后向该集群提交Flink任务。
2>单作业模式(Per-Job Mode)：一个作业一个集群，作业之间相互隔离。

切换到flink安装目录执行提交作业命令：cd /usr/local/apps/flink-1.13.5

以下是flink作业提交命令参数说明：

./bin/flink run
-m yarn-cluster #运行模式
-d                     #如果存在，则以分离的方式运行作业模式
-yjm 1024        #为jobmanager分配内存
-ytm 1024       #为taskmanager分配内存
-ynm Alert      #提交到Yarn上程序的名字
-ys 2               #ys表示taskslot数量
-yqu root.flink    #提交到指定队列
-c com.****.Alert /usr/local/apps/flinkjob/alert.jar  #指定执行主类名称和执行主类的路径
-path /usr/local/apps/flinkjob/alert.properties       #读取配置文件路径