前提条件
拥有1台CentOS7
CentOS7安装好jdk，官方文档要求java 11，使用java 8也可以。可参考 CentOS7安装jdk8
下载安装包
下载安装包
[hadoop@node1 ~]$ cd installfile/ [hadoop@node1 installfile]$ wget https://archive.apache.org/dist/flink/flink-1.17.1/flink-1.17.1-bin-scala_2.12.tgz --no-check-certificate
解压安装包
[hadoop@node1 installfile]$ tar -zxvf flink-1.17.1-bin-scala_2.12.tgz -C ~/soft
创建软链接
[hadoop@node1 installfile]$ cd ~/soft/ [hadoop@node1 soft]$ ls flink-1.18.0 hadoop hadoop-2.7.3 hbase hbase-1.7.1 jdk jdk1.8.0_271 [hadoop@node1 soft]$ ln -s flink-1.18.0 flink [hadoop@node1 soft]$ ls flink flink-1.18.0 hadoop hadoop-2.7.3 hbase hbase-1.7.1 jdk jdk1.8.0_271
设置环境变量
[hadoop@node1 soft]$ nano ~/.bashrc
添加如下环境变量
export FLINK_HOME=~/soft/flink export PATH=$PATH:$FLINK_HOME/bin
让环境变量立即生效
[hadoop@node1 soft]$ source ~/.bashrc
验证版本号
[hadoop@node1 soft]$ flink -v SLF4J: Class path contains multiple SLF4J bindings. SLF4J: Found binding in [jar:file:/home/hadoop/soft/flink-1.17.1/lib/log4j-slf4j-impl-2.17.1.jar!/org/slf4j/impl/StaticLoggerBinder.class] SLF4J: Found binding in [jar:file:/home/hadoop/soft/hadoop-2.7.3/share/hadoop/common/lib/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class] SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation. SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory] Version: 1.17.1, Commit ID: 2750d5c
配置flink
进入flink配置目录，查看flink的配置文件
[hadoop@node1 soft]$ cd $FLINK_HOME/ [hadoop@node1 flink]$ ls bin conf examples lib LICENSE licenses log NOTICE opt plugins README.txt [hadoop@node1 flink]$ cd conf/ [hadoop@node1 conf]$ ls flink-conf.yaml log4j-console.properties log4j-session.properties logback-session.xml masters zoo.cfg log4j-cli.properties log4j.properties logback-console.xml logback.xml workers
配置flink-conf.yaml
[hadoop@node1 conf]$ vim flink-conf.yaml
找到如下配置项，并按照如下修改，其中node1为机器主机名，请根据实际修改。
jobmanager.rpc.address: node1 jobmanager.bind-host: 0.0.0.0 taskmanager.bind-host: 0.0.0.0 taskmanager.host: node1 rest.address: node1 rest.bind-address: 0.0.0.0
配置master
[hadoop@node1 conf]$ nano masters
修改masters内容为
node1:8081
配置workers
[hadoop@node1 conf]$ vim masters
修改workers内容为
node1
查看flink的命令
[hadoop@node1 conf]$ cd $FLINK_HOME/bin [hadoop@node1 bin]$ ls bash-java-utils.jar flink-daemon.sh kubernetes-taskmanager.sh start-cluster.sh yarn-session.sh config.sh historyserver.sh pyflink-shell.sh start-zookeeper-quorum.sh zookeeper.sh find-flink-home.sh jobmanager.sh sql-client.sh stop-cluster.sh flink kubernetes-jobmanager.sh sql-gateway.sh stop-zookeeper-quorum.sh flink-console.sh kubernetes-session.sh standalone-job.sh taskmanager.sh
集群启动与停止
启动flink集群
[hadoop@node1 bin]$ start-cluster.sh Starting cluster. Starting standalonesession daemon on host node1. Starting taskexecutor daemon on host node1.
查看进程
[hadoop@node1 bin]$ jps 2433 TaskManagerRunner 2469 Jps 2143 StandaloneSessionClusterEntrypoint
Web UI
浏览器访问
ip:8081
或者
主机名称:8081
注意：如果用windows的浏览器访问，需要先在windows的hosts文件添加ip和主机名node1的映射。
关闭flink集群
[hadoop@node1 bin]$ stop-cluster.sh Stopping taskexecutor daemon (pid: 2433) on host node1. Stopping standalonesession daemon (pid: 2143) on host node1.
查看进程
[hadoop@node1 bin]$ jps 3239 Jps
单独启动flink进程
[hadoop@node1 bin]$ jobmanager.sh start [hadoop@node1 bin]$ taskmanager.sh start
操作过程如下
[hadoop@node1 bin]$ jobmanager.sh start Starting standalonesession daemon on host node1. [hadoop@node1 bin]$ jps 3522 StandaloneSessionClusterEntrypoint 3593 Jps [hadoop@node1 bin]$ taskmanager.sh start Starting taskexecutor daemon on host node1. [hadoop@node1 bin]$ jps 3522 StandaloneSessionClusterEntrypoint 3878 TaskManagerRunner 3910 Jps
单独关闭flink进程
[hadoop@node1 bin]$ taskmanager.sh stop [hadoop@node1 bin]$ jobmanager.sh stop
操作过程如下
[hadoop@node1 bin]$ taskmanager.sh stop Stopping taskexecutor daemon (pid: 3878) on host node1. [hadoop@node1 bin]$ jps 3522 StandaloneSessionClusterEntrypoint 4244 Jps [hadoop@node1 bin]$ jobmanager.sh stop Stopping standalonesession daemon (pid: 3522) on host node1. [hadoop@node1 bin]$ jps 4536 Jps
提交应用测试
启动flink集群
[hadoop@node1 flink]$ start-cluster.sh
提交flink作业
运行flink提供的wordcount案例程序
[hadoop@node1 flink]$ flink run examples/streaming/WordCount.jar SLF4J: Class path contains multiple SLF4J bindings. SLF4J: Found binding in [jar:file:/home/hadoop/soft/flink-1.17.1/lib/log4j-slf4j-impl-2.17.1.jar!/org/slf4j/impl/StaticLoggerBinder.class] SLF4J: Found binding in [jar:file:/home/hadoop/soft/hadoop-2.7.3/share/hadoop/common/lib/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class] SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation. SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory] Executing example with default input data. Use --input to specify file input. Printing result to stdout. Use --output to specify output path. Job has been submitted with JobID 20126e9c38bfaf5651db0f068e78e37a Program execution finished Job with JobID 20126e9c38bfaf5651db0f068e78e37a has finished. Job Runtime: 1999 ms
查看结果
查看输出的wordcount结果的末尾10行数据
[hadoop@node1 flink]$ tail log/flink-*-taskexecutor-*.out (nymph,1) (in,3) (thy,1) (orisons,1) (be,4) (all,2) (my,1) (sins,1) (remember,1) (d,4)
Web UI查看提交的作业
Web UI查看作业结果
参考：First steps | Apache Flink
完成！enjoy it!