如果是Flink 1.11 以上，建议用yarn-application 模式。

Flink的Yarn模式会在Yarn集群中动态创建一个Flink Cluster，然后你就可以往这个Flink Session Cluster提交Flink Job了。

配置

  • 除了配置 FLINK_HOME 和 flink.execution.mode 外还需要配置 HADOOP_CONF_DIR，并且要确保Zeppelin这台机器可以访问你的hadoop集群, 你可以运行hadoop fs 命令来验证是否能够连接hadoop集群。

image.png

  • 确保Zeppelin这台机器上安装了hadoop客户端（就是hadoop的那些jar包和配置文件），并且hadoop命令在环境变量 PATH上。因为Flink本身不带hadoop相关依赖（你可以把flink-shaded-hadoop 拷贝到flink lib下），所以flink的推荐方式是在提交Flink作业的时候运行hadoop classpath命令，然后把hadoop相关的jar放到CLASSPATH上。

Interpreter binding 模式

如果是Shared 模式，所有的job都会运行一个Flink Yarn Session Cluster里。如果是Per User Isolated，那么每个用户都会运行一个Flink Yarn Session Cluster，如果是Per Note Isolated，那么每个Note都对运行一个Flink Yarn Session Cluster。

Flink Cluster 生命周期

Yarn 模式的 Flink Cluster 就是上文说的运行Yarn集群里的Yarn Session Cluster，他的生命周期是Zeppelin控制的。你启动Flink Interpreter的时候就创建这个Yarn Session Cluster，当你停止或者重启Flink Interpreter的时候就会销毁这个Yarn Session Cluster。

配置 Yarn Session Cluster

配置项 描述
flink.yarn.appName yarn app name
flink.jm.memory JobManager的内存大小，单位是M，默认是1024
flink.tm.memory TaskManager的内存大熊啊，单位是M，默认是1024
flink.tm.slot 每个TaskManager的slot数目
flink.yarn.queue Yarn application queue

视频教程

Flink on Zeppelin 5. Yarn模式.mp4 (33.36MB)

钉钉群+公众号

Flink on Zeppelin 3群钉钉.JPG image.png