flink job 提交模式

文章目录

• flink job 提交模式
• • 1 maven 依赖引用
  • 2 提交模式
  • • 2.1 Standalone 模式
    • • 2.1.1 页面手动提交
      • 2.1.2 指令提交
      • 2.1.3 指令取消job
  • 2.2 yarn-会话模式(推荐)
  • • 2.2.1 准备
    • 2.2.2 步骤
    • • 2.2.2.1 启动集群
      • 2.2.2.2 提交作业
    • 2.2.3单作业部署提交

flink job 提交模式

flink版本：1.13.1
scala版本：2.12

1 maven 依赖引用

    <properties>
        <flink.version>1.13.1</flink.version>
        <scala.version>2.12</scala.version>
    </properties>
    
    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner-blink_${scala.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge_${scala.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!-- 实现自定义的数据格式来做序列化，可以引入下面的依赖 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-common</artifactId>
            <version>${flink.version}</version>
        </dependency> 
    </dependencies>

2 提交模式 2.1 Standalone 模式 2.1.1 页面手动提交

2.1.2 指令提交

# flink run -c 启动方法 -p 并行度 上传jar目录 参数,-d 标识后端运行
./bin/flink run -d -c class.main -p 3 xxx.jar args

2.1.3 指令取消job

# 1. 列出当前的job flink list -a 后缀添加-a表示查看状态
./bin/flink list
# 2. 取消job flink cancel jobId
./bin/flink cancel jobId

2.2 yarn-会话模式(推荐) 2.2.1 准备

1. hadoop集群环境搭建：https://blog.csdn.net/qq_27242695/article/details/120169968
2. flink 安装配置环境，支持hadoop yarn
   /ect/profile.d/下创建haddoode_env.sh文件

#HADOOD_HOME
export JAVA_HOME=/opt/module/jdk1.8.0_301
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`

2.2.2 步骤 2.2.2.1 启动集群

1. 启动 hadoop 集群(HDFS, YARN)。
2. 执行脚本命令向 YARN 集群申请资源，开启一个 YARN 会话，启动 Flink 集群。

bin/yarn-session.sh -nm test -d

• -d：分离模式，如果你不想让 Flink YARN 客户端一直前台运行，可以使用这个参数，即使关掉当前对话窗口，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 的会话模式也不会把集群资源固定，同样是动态分配的。

4. 访问flink集群控制端

5 hadoop 控制端

2.2.2.2 提交作业

1. 执行命令提交作业

• 将 Standalone 模式讲解中打包好的任务运行 JAR 包上传至集群
• 执行以下命令将该任务提交到已经开启的 Yarn-Session 中运行。

 flink run -c com.flink.Test /opt/module/flink-1.13.1/examples/table/flink-1.13.1.jar 

客户端可以自行确定 JobManager 的地址，也可以通过-m 或者-jobmanager 参数指定
JobManager 的地址，JobManager 的地址在 YARN Session 的启动页面中可以找到。

2. 查看及取消集群job

# 查看集群job
flink list
# 取消集群job
flink cancel 3c0c9aa56bd233ecd003df498dbc2d69

2.2.3单作业部署提交

即提交一个job启动一个集群

• 新版本提交指令

bin/flink run -d -t yarn-per-job -c com.class.main FlinkTutorial-1.0-SNAPSHOT.jar

• 早期版本

bin/flink run -m yarn-cluster -c com.class.main FlinkTutorial-1.0-SNAPSHOT.jar

• 可以使用命令行查看或取消作业，命令如下

$ ./bin/flink list -t yarn-per-job -Dyarn.application.id=application_XXXX_YY
$ ./bin/flink cancel -t yarn-per-job -Dyarn.application.id=application_XXXX_YY <jobId>