Flink Yarn-Session Job 相关常用脚本-恒创新域建站

★ FEATURED ARTICLE

一、Flink Yarn-Session 自动列出 session 内所有 running job批量 stop脚本1. 重要前置提醒脚本只做预览批量 stop务必先试运行看输出确认 jobId 无误再执行真实 stop仅适用于Yarn SessionFlink 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 CMDflink 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.sh2第一步预览强烈必做sh flink_session_batch_stop.sh application_1756000000000_0001 hdfs:///flink/savepoints输入Y只会打印 jobId 和 stop 命令不会真正停止任务核对 job 是否是你想要下线的。3第二步真实执行 stop预览确认无误后再次执行脚本输入N脚本会逐个执行 flink stopsh flink_session_batch_stop.sh application_1756000000000_0001 hdfs:///flink/savepoints4所有 Job stop 完成之后关闭 Session释放常驻 TM 容器flink cancel -t yarn-session -Dyarn.application.idapplication_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留人工确认环节规避生产风险。五、DN/NM退役整套缩容操作顺序脚本串联版flink_session_batch_stop.sh→ 预览确认 Job执行批量 stopflink_session_check_job.sh→ 校验确认无 RUNNING job人工确认后flink cancel关闭 Session释放 TM 容器yarn node -status xxx:8041确认容器数 0继续执行 NM 黑名单、DN 退役流程

阅读完成 · 觉得有帮助？

咨询建站