Flink架构图:
一、环境要求
1、JDK1.8+（JAVA_HOME环境变量设置，也可以在conf/flink-conf.yaml设置变量env.java.home）
2、SSH免密
二、下载安装(Standalone Cluster HA)
1、下载选择
根据需要的flink版本及对应已有的hadoop版本进行下载 ，本次实操环境选择flink1.6，hadoop2.7 下载页面
2、解压安装
wget http://archive.apache.org/dist/flink/flink-1.6.2/flink-1.6.2-bin-hadoop27-scala_2.11.tgz
tar xvf flink-1.6.2-bin-hadoop27-scala_2.11.tgz -C /opt/
ln -s /opt/flink-1.6.2/ /opt/apps/flink
3、修改配置
这里不使用flink自带的zookeeper,conf/zoo.cfg可以不用理会。
修改conf/flink-conf.yaml
还可通过设置 jobmanager.heap.mb 和 taskmanager.heap.mb 的值来定义JVM最大内存量（单位为MB）
# 设置为 master 节点
jobmanager.rpc.address: hdc-data1
# 配置高可用
high-availability: zookeeper
high-availability.zookeeper.quorum: hdc-data1:2181,hdc-data2:2181,hdc-data3:2181
high-availability.storageDir: hdfs:///flink/recovery
# 添加ZooKeeper根节点，在该节点下放置所有集群节点
high-availability.zookeeper.path.root: /flink
# 添加ZooKeeper 的 cluster-id 节点，在该节点下放置集群的所有相关数据
#注意：flink on yarn 模式下不建议设置high-availability.cluster-id参数，因为会造成所有yarn-session指向同一个flink集群。在不设置该参数的情况下将以yarn applictionid作为clusterid
#high-availability.cluster-id: /flink_cluster
#【可选】状态保存
state.checkpoints.dir: hdfs:///flink-checkpoints
state.savepoints.dir: hdfs:///flink-checkpoints
#【可选】修改默认端口
rest.port: 8181
#当指定该参数时rest.port将不生效，这样在多个yarn-session下就不会造成端口冲突
rest.bind-port: 50100-50200
historyserver.web.port: 8182
#【可选】配置historyserver
jobmanager.archive.fs.dir: hdfs:///flink/completed-jobs/
historyserver.archive.fs.dir: hdfs:///flink/completed-jobs/
在conf/masters配置masters节点，包含启动 JobManagers 的所有主机以及 Web 用户界面绑定的端口
hdc-data1:8181
hdc-data2:8181
在conf/slaves配置worker节点
hdc-data2
hdc-data3
4、启动集群
1、先启动hdfs和zookeeper服务
2、启动flink集群
bin/start-cluster.sh
3、ui查看
active
standy
三、Yarn Cluster HA
在standalone的基础上添加yarn配置支持。
1、在yarn安装配置yarn-site.xml中添加yarn最大重试次数，默认2次。
<property>
<name>yarn.resourcemanager.am.max-attempts</name>
<value>4</value>
<description>
The maximum number of application master execution attempts.
</description>
</property>
2、必须在flink-conf.yaml中也配置最大重试次数，添加如下参数：
yarn.application-attempts: 10
注意：yarn.resourcemanager.am.max-attempts是应用程序重新启动的上限。这意味着在如果程序启动失败，YARN会再重试9次（9 次重试 + 1次启动），如果启动10次作业还失败，yarn才会将该任务的状态置为失败。如果抢占，节点硬件故障或重启，NodeManager 重新同步等操作需要，YARN继续尝试启动应用。这些重启尝试不计入 yarn.application-attempts个数中，请参阅 Jian Fang的博客。重要的是要注意 yarn.resourcemanager.am.max-attempts 为yarn中程序重启上限。因此，Flink 中设置的程序尝试次数不能超过启动 YARN 的集群设置。
3、yarn-session启动集群及提交任务
# ./bin/yarn-session.sh -h
Usage:
Required
-n,--container <arg> 为YARN分配容器的数量 (=Number of Task Managers)
Optional
-D <property=value> 动态属性
-d,--detached 以分离模式运行作业
-h,--help Yarn session帮助.
-id,--applicationId <arg> 连接到一个正在运行的YARN session
-j,--jar <arg> Flink jar文件的路径
-jm,--jobManagerMemory <arg> JobManager的内存大小，driver-memory [in MB]
-m,--jobmanager <arg> Address of the JobManager (master) to which to connect. Use this flag to connect to a different JobManager than the one specified in the configuration.
-n,--container <arg> TaskManager的数量，相当于executor的数量
-nm,--name <arg> 设置YARN应用自定义名称
-q,--query 显示可用的YARN资源 (memory, cores)
-qu,--queue <arg> 指定YARN队列
-s,--slots <arg> 每个JobManager的core的数量，executor-cores。建议将slot的数量设置每台机器的处理器数量
-st,--streaming 在流模式下启动Flink
-t,--ship <arg> 在指定目录中传送文件(t for transfer)
-tm,--taskManagerMemory <arg> 每个TaskManager的内存大小，executor-memory [in MB]
-yd,--yarndetached 如果存在，则以分离模式运行作业 (deprecated; use non-YARN specific option instead)
-z,--zookeeperNamespace <arg> 为高可用性模式创建Zookeeper子路径的命名空间
example( 以分离模式运行)
./bin/yarn-session.sh -n 3 -jm 1024 -tm 1024 -s 3 -nm FlinkOnYarnSession -d -st
提交作业
./bin/flink run -yid xxxx -c org.apache.flink.examples.java.wordcount.WordCount examples/batch/WordCount.jar
当暂停jod，可通过以下命令重新启动
./bin/yarn-session.sh -id application_id
