一、前期准备1、查看网卡2、配置静态IPvi /etc/sysconfig/network-scripts/ifcfg-ens32 ---- 根据自己网卡设置。3、设置主机名hostnamectl --static set-hostname 主机名例如hostnamectl --static set-hostname hadoop0014、配置IP与主机名映射vi /etc/hosts5、关闭防火墙systemctl stop firewalldsystemctl disable firewalld6、配置免密登录传送门二、JDK的安装传送门注意Flink1.16.0版本也支持使用JDK8后续版本对JDK8的支持将会移除。从Flink 1.17.0版本开始必须使用Java 11或更高版本来运行Flink。这是因为Flink为了支持最新的Java API和语言特性需要Java 11中引入的一些新功能。虽然使用JDK8 也可以但从 Flink 1.17 开始部分依赖于 Flink 的第三方库已经弃用了对 JDK 8 的支持并要求使用 JDK 11 或更高版本。考虑到Flink后期与一些大数据框架进行整合这些大数据框架对JDK11的支持并不完善例如Hive3.1.3版本还不支持JDK11所以采用JDK8来开发Flink。三、Flink的本地安装1、Flink的下载安装1.1. 下载Index of /dist/flink/flink-1.17.0https://archive.apache.org/dist/flink/flink-1.17.0/flink-1.17.0-bin-scala_2.12.tgz下载 flink-1.17.0-bin-scala_2.12.tgz 安装包1.2 上传使用xshell上传到指定安装路径此处是安装路径是 /opt/module1.3 解压重命名tar -zxvf flink-1.17.0-bin-scala_2.12.tgzmv flink-1.17.0 flink1.4 配置环境变量vi /etc/profileexport JAVA_HOME/opt/module/javaexport CLASSPATH.:\$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jarexport FLINK_HOME/opt/module/flinkexport PATH$PATH:\$JAVA_HOME/bin:$FLINK_HOME/bin1.5 加载环境变量source /etc/profile验证环境变量是否生效:env | grep HOMEenv | grep PATH2、修改flink-conf.yaml配置文件cd /opt/module/flinkvi conf/flink-conf.yaml# JobManager节点地址jobmanager.rpc.address: hadoop001jobmanager.bind-host: 0.0.0.0rest.address: hadoop001rest.bind-address: 0.0.0.0# TaskManager节点地址.需要配置为当前机器名taskmanager.bind-host: 0.0.0.0taskmanager.host: hadoop0013、启动集群cd /opt/module/flink./bin/start-cluster.sh4、查看进程5、查看WebUI默认端口 8081http://192.168.200.151:8081/6、停止集群cd \$FLINK_HOME./bin/stop-cluster.sh三、Flink Standalone搭建准备三台节点hadoop001、hadoop002、hadoop003每个节点部署不同角色。进程节点Master(JobManager)Slave(TaskManager)hadoop001Slave(TaskManager)hadoop002Slave(TaskManager)hadoop003以下内容是在Flink本地安装的基础上进行的。1、修改workers文件cd \$FLINK_HOMEcat \$FLINK_HOME/conf/workers EOFhadoop001hadoop002hadoop003EOF2、分发文件scp -r/etc/profile roothadoop002:/etc/profilescp -r/etc/profile roothadoop003:/etc/profilescp -r/opt/module/java roothadoop002:/opt/module/javascp -r/opt/module/java roothadoop003:/opt/module/javascp -r/opt/module/flink roothadoop002:/opt/module/flinkscp -r/opt/module/flink roothadoop003:/opt/module/flink让三台机器文件生效ssh hadoop001 source /etc/profilessh hadoop002 source /etc/profilessh hadoop003 source /etc/profile3、启动集群cd \$FLINK_HOME./bin/start-cluster.sh4、查看WebUI默认端口 8081http://192.168.200.151:8081/5、测试尝试提交一个简单任务如果任务正常执行完毕则集群一切正常。提交Flink自带的简单任务如下cd /opt/module/flink./bin/flink run examples/streaming/WordCount.jar6、停止集群cd /opt/module/flink./bin/stop-cluster.sh四、Flink Standalone HA搭建不推荐后面由Yarn管理HA的主要作用可以在集群中启动多个JobManager并使它们都向ZooKeeper进行注册ZooKeeper利用自身的选举机制保证同一时间只有一个JobManager是活动状态Active的其他的都是备用状态Standby。当活动状态的JobManager出现故障时ZooKeeper会从其他备用状态的JobManager选出一个成为活动JobManager进程节点Master(JobManager)Slave(TaskManager)FlinkZooKeeperQuorumPeerhadoop001Master(JobManager)Slave(TaskManager)FlinkZooKeeperQuorumPeerhadoop002Slave(TaskManager)FlinkZooKeeperQuorumPeerhadoop003下面内容在 Flink Standalone 搭建前提下修改。1、修改flink-conf.yaml配置文件cd /opt/module/flinkvi conf/flink-conf.yaml# 将高可用模式设置为ZooKeeper默认集群不会开启高可用状态high-availability: zookeeper# ZooKeeper集群主机名或IP与端口列表多个以逗号分隔high-availability.zookeeper.quorum: hadoop001:2181,hadoop002:2181,hadoop003:2181# 用于持久化JobManager元数据JobGraph、应用程序JAR文件等的HDFS地址以便进行故障恢复ZooKeeper上存储的只是元数据所在的位置路径信息high-availability.storageDir: /opt/module/flink/ha2、修改master文件cd /opt/module/flinkvi conf/mastershadoop001:8081hadoop002:80823、分发文件scp -r$FLINK_HOME/conf/flink-conf.yaml roothadoop002:$FLINK_HOME/conf/flink-conf.yamlscp -r$FLINK_HOME/conf/flink-conf.yaml roothadoop003:$FLINK_HOME/conf/flink-conf.yamlscp -r$FLINK_HOME/conf/masters roothadoop002:$FLINK_HOME/conf/mastersscp -r$FLINK_HOME/conf/masters roothadoop003:$FLINK_HOME/conf/masters4、安装与启动Zookeeper传送门5、启动集群cd /opt/module/flink./bin/start-cluster.sh6、查看WebUI默认端口 8081http://192.168.200.151:8081/7、测试尝试提交一个简单任务如果任务正常执行完毕则集群一切正常。提交Flink自带的简单任务如下cd /opt/module/flink./bin/flink run examples/streaming/WordCount.jar8、停止集群cd /opt/module/flink./bin/stop-cluster.sh五、YARN运行模式重点YARN上部署的过程是客户端把Flink应用提交给Yarn的ResourceManagerYarn的ResourceManager会向Yarn的NodeManager申请容器。在这些容器上Flink会部署JobManager和TaskManager的实例从而启动集群。Flink会根据运行在JobManger上的作业所需要的Slot数量动态分配TaskManager资源。本节是在三、Flink Standalone搭建的基础上进行修改。1、安装启动Hadoop集群在将Flink任务部署至YARN集群之前需要确认集群是否安装有Hadoop保证Hadoop版本至少在2.2以上并且集群中安装有HDFS服务。传送门2、配置环境变量vi /etc/profileexport HADOOP_CONF_DIR${HADOOP_HOME}/etc/hadoopexport HADOOP_CLASSPATHhadoop classpath不要丢掉 一漂符号“”注意重点是HADOOP_CLASSPATHhadoop classpath 是执行命令 hadoop classpath使环境变量能够加载到hadoop的类路径和包路径。此时可以将hadoop和Flink进行解耦不用纠结使用hadoop的哪个版本。3、修改配置文件flink-conf.yaml1.此文件中的 hadoop001 无需修改启动时候yarn自动分配代理的主机名和端口# JobManager节点地址.jobmanager.rpc.address:hadoop001jobmanager.bind-host: 0.0.0.0rest.address: hadoop001rest.bind-address: 0.0.0.0# TaskManager节点地址.需要配置为当前机器名taskmanager.bind-host: 0.0.0.0taskmanager.host:hadoop0012、设置加载检查vi /opt/module/flink/conf/flink-conf.yamlclassloader.check-leaked-classloader: false注意如果上面配置中是避免启动过程中报如下异常。Exception in thread “Thread-5” java.lang.IllegalStateException: Trying to access closed classloader. Please check if you store classloaders directly or indirectly in static fields. If the stacktrace suggests that the leak occurs in a third party library and cannot be fixed immediately, you can disable this check with the configuration ‘classloader.check-leaked-classloader’.at org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders4、分发文件scp -r/etc/profile roothadoop002:/etc/profilescp -r/etc/profile roothadoop003:/etc/profilescp -r/opt/module/flink/conf/flink-conf.yamlroothadoop002:/opt/module/flink/conf/flink-conf.yamlscp -r/opt/module/flink/conf/flink-conf.yamlroothadoop003:/opt/module/flink/conf/flink-conf.yaml三台机器分别执行 source /etc/profile5、Session会话模式YARN的会话模式与独立集群略有不同需要首先申请一个YARN会话YARN Session来启动Flink集群。5.1.启动Hadoop集群cd /opt/module/hadoopsbin/start-all.sh5.2.启动Flink的会话模式cd /opt/module/flinkbin/yarn-session.sh -d -nm flinkTest参数说明-d分离模式执行命令后不会占用窗口即使关掉当前对话窗口YARN session也可以后台运行。-jm--jobManagerMemory配置JobManager所需内存默认单位MB。-nm--name配置在YARN UI界面上显示的任务名。-qu--queue指定YARN队列名。-tm--taskManager配置每个TaskManager所使用内存。注意Flink1.11.0版本不再使用-n参数和-s参数分别指定TaskManager数量和slot数量YARN会按照需求动态分配TaskManager和slot。所以从这个意义上讲YARN的会话模式也不会把集群资源固定同样是动态分配的。YARN Session启动之后会给出一个Web UI地址以及一个YARN application ID如下所示此时Yarn为Flink动态分配资源并启动JobManager。用户可以通过Web UI或者命令行两种方式提交作业。5.3.命令行提交作业新打开一个窗口使用命令行方式提交作业提交后Yarn会自动的为Flink分配资源启动对应的TaskManager。运行自己到 webui上进行查看。cd /opt/module/flink./bin/flink run examples/streaming/WordCount.jar5.4.小结通过前面来看在yarn的每个应用其实就是对应 flink 的 一个集群6、单作业模式前面不变只是提交方式改变。cd/opt/module/flinkbin/flink run -d -t yarn-per-job examples/streaming/WordCount.jar如果报错如图所示则配置flink-conf.yamlvi /opt/module/flink/conf/flink-conf.yamlenv.java.home: /opt/module/javaenv.hadoop.conf.dir: /opt/module/hadoop/etc/hadoop修改后分发文件scp -r/opt/module/flink/conf/flink-conf.yamlroothadoop002:/opt/module/flink/conf/flink-conf.yamlscp -r/opt/module/flink/conf/flink-conf.yamlroothadoop003:/opt/module/flink/conf/flink-conf.yaml7、应用模式前面不变只是提交方式改变。cd/opt/module/flinkbin/flink run-application -t yarn-application examples/streaming/WordCount.jar如果 Flink 本身的依赖和j插件的jar用户可以预先上传到HDFS而不需要每次单独发送到集群这就使得作业提交更加轻量了。1、创建HDFS目录hdfs dfs -mkdir /flink-disthdfs dfs -mkdir /flink-jars2、上传 Flink 本身的依赖和用户jarhdfs dfs -put /opt/module/flink/lib/ /flink-disthdfs dfs -put /opt/module/flink/plugins/ /flink-disthdfs dfs -put /opt/module/flink/examples/streaming/WordCount.jar /flink-dist3、提交作业cd/opt/module/flinkbin/flink run-application -t yarn-application -Dyarn.provided.lib.dirshdfs://hadoop001:9000/flink-dist hdfs://hadoop001:9000/flink-jars/WordCount.jar 一、前期准备1、查看网卡2、配置静态IPvi /etc/sysconfig/network-scripts/ifcfg-ens32 ---- 根据自己网卡设置。3、设置主机名hostnamectl --static set-hostname 主机名例如hostnamectl --static set-hostname hadoop0014、配置IP与主机名映射vi /etc/hosts5、关闭防火墙systemctl stop firewalldsystemctl disable firewalld6、配置免密登录传送门二、JDK的安装传送门注意Flink1.16.0版本也支持使用JDK8后续版本对JDK8的支持将会移除。从Flink 1.17.0版本开始必须使用Java 11或更高版本来运行Flink。这是因为Flink为了支持最新的Java API和语言特性需要Java 11中引入的一些新功能。虽然使用JDK8 也可以但从 Flink 1.17 开始部分依赖于 Flink 的第三方库已经弃用了对 JDK 8 的支持并要求使用 JDK 11 或更高版本。考虑到Flink后期与一些大数据框架进行整合这些大数据框架对JDK11的支持并不完善例如Hive3.1.3版本还不支持JDK11所以采用JDK8来开发Flink。三、Flink的本地安装1、Flink的下载安装1.1. 下载Index of /dist/flink/flink-1.17.0https://archive.apache.org/dist/flink/flink-1.17.0/flink-1.17.0-bin-scala_2.12.tgz下载 flink-1.17.0-bin-scala_2.12.tgz 安装包1.2 上传使用xshell上传到指定安装路径此处是安装路径是 /opt/module1.3 解压重命名tar -zxvf flink-1.17.0-bin-scala_2.12.tgzmv flink-1.17.0 flink1.4 配置环境变量vi /etc/profileexport JAVA_HOME/opt/module/javaexport CLASSPATH.:\$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jarexport FLINK_HOME/opt/module/flinkexport PATH$PATH:\$JAVA_HOME/bin:$FLINK_HOME/bin1.5 加载环境变量source /etc/profile验证环境变量是否生效:env | grep HOMEenv | grep PATH2、修改flink-conf.yaml配置文件cd /opt/module/flinkvi conf/flink-conf.yaml# JobManager节点地址jobmanager.rpc.address: hadoop001jobmanager.bind-host: 0.0.0.0rest.address: hadoop001rest.bind-address: 0.0.0.0# TaskManager节点地址.需要配置为当前机器名taskmanager.bind-host: 0.0.0.0taskmanager.host: hadoop0013、启动集群cd /opt/module/flink./bin/start-cluster.sh4、查看进程5、查看WebUI默认端口 8081http://192.168.200.151:8081/6、停止集群cd \$FLINK_HOME./bin/stop-cluster.sh三、Flink Standalone搭建准备三台节点hadoop001、hadoop002、hadoop003每个节点部署不同角色。进程节点Master(JobManager)Slave(TaskManager)hadoop001Slave(TaskManager)hadoop002Slave(TaskManager)hadoop003以下内容是在Flink本地安装的基础上进行的。1、修改workers文件cd \$FLINK_HOMEcat \$FLINK_HOME/conf/workers EOFhadoop001hadoop002hadoop003EOF2、分发文件scp -r/etc/profile roothadoop002:/etc/profilescp -r/etc/profile roothadoop003:/etc/profilescp -r/opt/module/java roothadoop002:/opt/module/javascp -r/opt/module/java roothadoop003:/opt/module/javascp -r/opt/module/flink roothadoop002:/opt/module/flinkscp -r/opt/module/flink roothadoop003:/opt/module/flink让三台机器文件生效ssh hadoop001 source /etc/profilessh hadoop002 source /etc/profilessh hadoop003 source /etc/profile3、启动集群cd \$FLINK_HOME./bin/start-cluster.sh4、查看WebUI默认端口 8081http://192.168.200.151:8081/5、测试尝试提交一个简单任务如果任务正常执行完毕则集群一切正常。提交Flink自带的简单任务如下cd /opt/module/flink./bin/flink run examples/streaming/WordCount.jar6、停止集群cd /opt/module/flink./bin/stop-cluster.sh四、Flink Standalone HA搭建不推荐后面由Yarn管理HA的主要作用可以在集群中启动多个JobManager并使它们都向ZooKeeper进行注册ZooKeeper利用自身的选举机制保证同一时间只有一个JobManager是活动状态Active的其他的都是备用状态Standby。当活动状态的JobManager出现故障时ZooKeeper会从其他备用状态的JobManager选出一个成为活动JobManager进程节点Master(JobManager)Slave(TaskManager)FlinkZooKeeperQuorumPeerhadoop001Master(JobManager)Slave(TaskManager)FlinkZooKeeperQuorumPeerhadoop002Slave(TaskManager)FlinkZooKeeperQuorumPeerhadoop003下面内容在 Flink Standalone 搭建前提下修改。1、修改flink-conf.yaml配置文件cd /opt/module/flinkvi conf/flink-conf.yaml# 将高可用模式设置为ZooKeeper默认集群不会开启高可用状态high-availability: zookeeper# ZooKeeper集群主机名或IP与端口列表多个以逗号分隔high-availability.zookeeper.quorum: hadoop001:2181,hadoop002:2181,hadoop003:2181# 用于持久化JobManager元数据JobGraph、应用程序JAR文件等的HDFS地址以便进行故障恢复ZooKeeper上存储的只是元数据所在的位置路径信息high-availability.storageDir: /opt/module/flink/ha2、修改master文件cd /opt/module/flinkvi conf/mastershadoop001:8081hadoop002:80823、分发文件scp -r$FLINK_HOME/conf/flink-conf.yaml roothadoop002:$FLINK_HOME/conf/flink-conf.yamlscp -r$FLINK_HOME/conf/flink-conf.yaml roothadoop003:$FLINK_HOME/conf/flink-conf.yamlscp -r$FLINK_HOME/conf/masters roothadoop002:$FLINK_HOME/conf/mastersscp -r$FLINK_HOME/conf/masters roothadoop003:$FLINK_HOME/conf/masters4、安装与启动Zookeeper传送门5、启动集群cd /opt/module/flink./bin/start-cluster.sh6、查看WebUI默认端口 8081http://192.168.200.151:8081/7、测试尝试提交一个简单任务如果任务正常执行完毕则集群一切正常。提交Flink自带的简单任务如下cd /opt/module/flink./bin/flink run examples/streaming/WordCount.jar8、停止集群cd /opt/module/flink./bin/stop-cluster.sh五、YARN运行模式重点YARN上部署的过程是客户端把Flink应用提交给Yarn的ResourceManagerYarn的ResourceManager会向Yarn的NodeManager申请容器。在这些容器上Flink会部署JobManager和TaskManager的实例从而启动集群。Flink会根据运行在JobManger上的作业所需要的Slot数量动态分配TaskManager资源。本节是在三、Flink Standalone搭建的基础上进行修改。1、安装启动Hadoop集群在将Flink任务部署至YARN集群之前需要确认集群是否安装有Hadoop保证Hadoop版本至少在2.2以上并且集群中安装有HDFS服务。传送门2、配置环境变量vi /etc/profileexport HADOOP_CONF_DIR${HADOOP_HOME}/etc/hadoopexport HADOOP_CLASSPATHhadoop classpath不要丢掉 一漂符号“”注意重点是HADOOP_CLASSPATHhadoop classpath 是执行命令 hadoop classpath使环境变量能够加载到hadoop的类路径和包路径。此时可以将hadoop和Flink进行解耦不用纠结使用hadoop的哪个版本。3、修改配置文件flink-conf.yaml1.此文件中的 hadoop001 无需修改启动时候yarn自动分配代理的主机名和端口# JobManager节点地址.jobmanager.rpc.address:hadoop001jobmanager.bind-host: 0.0.0.0rest.address: hadoop001rest.bind-address: 0.0.0.0# TaskManager节点地址.需要配置为当前机器名taskmanager.bind-host: 0.0.0.0taskmanager.host:hadoop0012、设置加载检查vi /opt/module/flink/conf/flink-conf.yamlclassloader.check-leaked-classloader: false注意如果上面配置中是避免启动过程中报如下异常。Exception in thread “Thread-5” java.lang.IllegalStateException: Trying to access closed classloader. Please check if you store classloaders directly or indirectly in static fields. If the stacktrace suggests that the leak occurs in a third party library and cannot be fixed immediately, you can disable this check with the configuration ‘classloader.check-leaked-classloader’.at org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders4、分发文件scp -r/etc/profile roothadoop002:/etc/profilescp -r/etc/profile roothadoop003:/etc/profilescp -r/opt/module/flink/conf/flink-conf.yamlroothadoop002:/opt/module/flink/conf/flink-conf.yamlscp -r/opt/module/flink/conf/flink-conf.yamlroothadoop003:/opt/module/flink/conf/flink-conf.yaml三台机器分别执行 source /etc/profile5、Session会话模式YARN的会话模式与独立集群略有不同需要首先申请一个YARN会话YARN Session来启动Flink集群。5.1.启动Hadoop集群cd /opt/module/hadoopsbin/start-all.sh5.2.启动Flink的会话模式cd /opt/module/flinkbin/yarn-session.sh -d -nm flinkTest参数说明-d分离模式执行命令后不会占用窗口即使关掉当前对话窗口YARN session也可以后台运行。-jm--jobManagerMemory配置JobManager所需内存默认单位MB。-nm--name配置在YARN UI界面上显示的任务名。-qu--queue指定YARN队列名。-tm--taskManager配置每个TaskManager所使用内存。注意Flink1.11.0版本不再使用-n参数和-s参数分别指定TaskManager数量和slot数量YARN会按照需求动态分配TaskManager和slot。所以从这个意义上讲YARN的会话模式也不会把集群资源固定同样是动态分配的。YARN Session启动之后会给出一个Web UI地址以及一个YARN application ID如下所示此时Yarn为Flink动态分配资源并启动JobManager。用户可以通过Web UI或者命令行两种方式提交作业。5.3.命令行提交作业新打开一个窗口使用命令行方式提交作业提交后Yarn会自动的为Flink分配资源启动对应的TaskManager。运行自己到 webui上进行查看。cd /opt/module/flink./bin/flink run examples/streaming/WordCount.jar5.4.小结通过前面来看在yarn的每个应用其实就是对应 flink 的 一个集群6、单作业模式前面不变只是提交方式改变。cd/opt/module/flinkbin/flink run -d -t yarn-per-job examples/streaming/WordCount.jar如果报错如图所示则配置flink-conf.yamlvi /opt/module/flink/conf/flink-conf.yamlenv.java.home: /opt/module/javaenv.hadoop.conf.dir: /opt/module/hadoop/etc/hadoop修改后分发文件scp -r/opt/module/flink/conf/flink-conf.yamlroothadoop002:/opt/module/flink/conf/flink-conf.yamlscp -r/opt/module/flink/conf/flink-conf.yamlroothadoop003:/opt/module/flink/conf/flink-conf.yaml7、应用模式前面不变只是提交方式改变。cd/opt/module/flinkbin/flink run-application -t yarn-application examples/streaming/WordCount.jar如果 Flink 本身的依赖和j插件的jar用户可以预先上传到HDFS而不需要每次单独发送到集群这就使得作业提交更加轻量了。1、创建HDFS目录hdfs dfs -mkdir /flink-disthdfs dfs -mkdir /flink-jars2、上传 Flink 本身的依赖和用户jarhdfs dfs -put /opt/module/flink/lib/ /flink-disthdfs dfs -put /opt/module/flink/plugins/ /flink-disthdfs dfs -put /opt/module/flink/examples/streaming/WordCount.jar /flink-dist3、提交作业cd/opt/module/flinkbin/flink run-application -t yarn-application -Dyarn.provided.lib.dirshdfs://hadoop001:9000/flink-dist hdfs://hadoop001:9000/flink-jars/WordCount.jar 
