2023年2月16日

Flink 集群部署

大数据 · Flink · 流处理

在三台服务器上部署 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

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

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_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_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`

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`

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 run -c word_count.无界流处理 ./JAR.jar

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: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: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

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

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，重新提交作业，应该就可以解决问题了。

X/Twitter

GitHub

d5v blog