Flink 集群部署

期间准备好 Jdk、hadoop 并启动hadoop

解压和设置flink环境变量

修改配置文件

vim $FLINK_HOME/conf/flink-conf.yaml

# jobmanager.rpc.address: localhost
# 改为
jobmanager.rpc.address: master

vim $FLINK_HOME/conf/masters
# 改为
master:8081

vim $FLINK_HOME/conf/workers
# 改为
slave1
slave2

启动 $FLINK_HOME/bin/start-cluster.sh

在哪个机器上运行这条命令 就会在那台机器上开放web端口

提交任务

$FLINK_HOME/bin/flink run -m master:8081 -c word_count.无界流处理 -p 2 /root/untitled1-1.0-SNAPSHOT-jar-with-dependencies.jar --host slave1 --port 7777

#$FLINK_HOME/bin/flink run -m HOST:PORT -c 入口类 -p 并行度 PATH/TO/JAR.jar --args value

正在运行的作业查看 flink list| flink list -a

关闭运行的作业 flink cancel <JOB_ID>

Flink的运行模式

独立模式 (Standalone)

独立模式包含以下部署方式:

• 单作业模式（借助其他管理平台如yarn、k8s）

• 会话模式

• 应用模式

YARN模式

将Flink应用提交给 Yarn的 ResourceManager

1、启动hadoop集群

2、检查环境变量是否有

export HADOOP_HOME=/opt/module/hadoop-3.1.3
export PATH=$PATH:$HADOOP_HOME/bin
export PATH=$PATH:$HADOOP_HOME/sbin
export HADOOP_CONF_DIR=${HADOOP_HOME}/etc/hadoop
export HADOOP_CLASSPATH=`hadoop classpath`

3、重启hadoop yarn

4、启动flink on yarn yarn-session.sh -nm test -d

参数:

• -nm <NAME>

• -d 分离模式

• -jm JobManager 所需内存 单位MB

• -qu 指定yarn队列名

• -tm 每个TaskManager 所所用内存

5、提交作业 flink run

flink run -c word_count.无界流处理 ./JAR.jar

其他：

Flink 运行报错

log4j:WARN No appenders could be found for logger (org.apache.flink.api.scala.ClosureCleaner$).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.

解决办法 修改log4j配置文件

新建log4j.properties文件

将其中的内容修改为以下内容：

log4j.rootLogger=ERROR, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target=System.out
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n

然后重启flink，重新提交作业，应该就可以解决问题了。