Flink Yarn-Session Job 相关常用脚本-尧图手机网站定制

一、Flink Yarn-Session 自动列出 session 内所有 running job，批量 stop脚本

1. 重要前置提醒

脚本只做预览 + 批量 stop，务必先试运行看输出，确认 jobId 无误再执行真实 stop
仅适用于Yarn Session，Flink on YARN
执行 stop 会生成 savepoint，提前确认 HDFS savepoint 目录有写入权限
生产环境：优先手动逐个 stop；脚本适合测试环境，或者 job 数量非常多的场景

2. shell 脚本：flink_session_batch_stop.sh

#!/bin/bash # 用法: sh flink_session_batch_stop.sh application_xxxx_0001 hdfs:///flink/savepoints # 接收2个入参：session的applicationId、savepoint路径 SESSION_APP_ID=$1 SAVEPOINT_PATH=$2 if [ $# -ne 2 ];then echo "用法：$0 <YarnSession ApplicationId> <savepoint hdfs路径>" echo "示例：$0 application_1756000000000_0001 hdfs:///flink/savepoints" exit 1 fi # 1. 列出session中所有RUNNING状态job，提取JobID echo "===== 正在查询Session内运行中的Job =====" JOB_IDS=$(flink list -t yarn-session -Dyarn.application.id=${SESSION_APP_ID} | grep "RUNNING" | awk '{print $4}') if [ -z "${JOB_IDS}" ];then echo "未查询到RUNNING的Job，无需停止" exit 0 fi echo "查到待停止JobID列表：" echo "${JOB_IDS}" echo "" # 2. 预览阶段：只打印命令，不执行真实stop（默认打开预览！） read -p "是否进入预览模式(只打印命令不执行)？输入N则真实执行stop [Y/N] " ANS for JOB_ID in ${JOB_IDS} do CMD="flink stop -p ${SAVEPOINT_PATH} ${JOB_ID}" echo ">>> JobID: ${JOB_ID}" echo " 执行命令：${CMD}" if [ "${ANS}" != "N" ] then echo " 【预览模式】不执行" else echo " 【真实执行】开始stop，等待生成savepoint..." ${CMD} if [ $? -eq 0 ];then echo " ${JOB_ID} stop成功，记住返回的savepoint地址！" else echo " ${JOB_ID} stop失败，请检查！" fi fi echo "-----------------------------------------" done echo "" echo "===== 全部Job处理完成 =====" echo "全部Job stop完成后，记得执行cancel关闭Session集群释放TM容器" echo "命令参考：flink cancel -t yarn-session -Dyarn.application.id=${SESSION_APP_ID} ${SESSION_APP_ID}"

3. 使用步骤

1）保存脚本并授权

vim flink_session_batch_stop.sh chmod +x flink_session_batch_stop.sh

2）第一步：预览（强烈必做！）

sh flink_session_batch_stop.sh application_1756000000000_0001 hdfs:///flink/savepoints

输入Y，只会打印 jobId 和 stop 命令，不会真正停止任务，核对 job 是否是你想要下线的。

3）第二步：真实执行 stop预览确认无误后再次执行脚本，输入N，脚本会逐个执行 flink stop：

sh flink_session_batch_stop.sh application_1756000000000_0001 hdfs:///flink/savepoints

4）所有 Job stop 完成之后，关闭 Session，释放常驻 TM 容器

flink cancel -t yarn-session -Dyarn.application.id=application_1756000000000_0001 application_1756000000000_0001

脚本说明 & 坑点

脚本通过grep RUNNING过滤，已经停止的 Job 不会被处理
flink stop 是同步阻塞操作，每一个 job 会等待 savepoint 生成完成，才会处理下一个 job，任务多的时候耗时久
脚本捕获返回码，如果某个 job stop 失败会打印错误，你可以介入排查
脚本不会自动保存输出的 savepoint 路径，需要手动记录每个 job 返回的 savepoint 地址，恢复任务要用

二、Flink Yarn-SessioFlink Yarn-Session Job 状态校验脚本n Job 状态校验脚本

作用：批量检查 Session 内 Job 状态，确认所有 RUNNING Job 都已经停止，用于执行完批量 stop 之后校验。

输出：当前运行中的 Job 列表；如果无 RUNNING Job，代表全部停止，可以继续 cancel Session。

#!/bin/bash # flink_session_check_job.sh # 用法：sh flink_session_check_job.sh <session applicationId> SESSION_APP_ID=$1 if [ $# -ne 1 ];then echo "用法：$0 <YarnSession ApplicationId>" echo "示例：$0 application_1756000000000_0001" exit 1 fi echo "===== 查询Session: ${SESSION_APP_ID} 内Job状态 =====" JOB_RAW=$(flink list -t yarn-session -Dyarn.application.id=${SESSION_APP_ID}) echo "${JOB_RAW}" # 提取RUNNING的JobID RUNNING_JOB=$(echo "${JOB_RAW}" | grep "RUNNING" | awk '{print $4}') echo -e "\n===== 校验结果 =====" if [ -z "${RUNNING_JOB}" ];then echo "校验通过：没有RUNNING状态Job，可以执行cancel Session" echo "执行命令参考：" echo "flink cancel -t yarn-session -Dyarn.application.id=${SESSION_APP_ID} ${SESSION_APP_ID}" else echo "校验不通过，还有正在运行的JobID：" echo "${RUNNING_JOB}" echo "请确认这些Job是否完成stop，不能cancel Session！" exit 1 fi

使用方法

# 授权 chmod +x flink_session_check_job.sh # 执行校验 sh flink_session_check_job.sh application_1756000000000_0001

输出两种情况

无 RUNNING Job：提示可以 cancel Session
还有运行 Job：打印 JobID，退出码非 0，阻止后续 cancel 操作

三、组合自动化脚本

把批量 stop + 状态校验合并，stop 全部任务后自动校验，校验通过提示 cancel session。 ⚠️ 生产不建议全自动 cancel，人工确认后再执行 cancel。

#!/bin/bash # flink_session_stop_all.sh SESSION_APP_ID=$1 SAVEPOINT_PATH=$2 if [ $# -ne 2 ];then echo "用法：$0 <session appId> <savepoint hdfs路径>" exit 1 fi #1. 获取running job JOB_IDS=$(flink list -t yarn-session -Dyarn.application.id=${SESSION_APP_ID} | grep "RUNNING" | awk '{print $4}') if [ -z "${JOB_IDS}" ];then echo "无运行Job" exit 0 fi echo "待停止Job列表：${JOB_IDS}" read -p "确认批量stop？[Y/N]" CONFIRM if [ "${CONFIRM}" != "Y" ];then echo "退出" exit 0 fi #2. 逐个stop for J in ${JOB_IDS} do echo "stop job ${J}" flink stop -p ${SAVEPOINT_PATH} ${J} done #3. 校验 echo -e "\n==== 开始校验Job状态 ====" RUNNING_JOB=$(flink list -t yarn-session -Dyarn.application.id=${SESSION_APP_ID} | grep "RUNNING" | awk '{print $4}') if [ -z "${RUNNING_JOB}" ];then echo "所有Job已停止，请手动执行cancel session释放TM" else echo "还有Job运行：${RUNNING_JOB}" exit 1 fi

四、要点

批量停止 Yarn Session 内 Job，我会写两段脚本：第一段批量 stop，并且增加预览模式，防止误杀； stop 执行完成后，调用校验脚本，查询 Session 内 Job 状态，确认不存在 RUNNING Job。只有校验通过，才可以手动 cancel Session，释放常驻 TM 容器。全程不自动 cancel Session，留人工确认环节，规避生产风险。