Flink任务自动恢复脚本

线上环境经常遇到flink任务挂掉得问题，这里写一个自动恢复脚本

# 我这里使用得datastream api编写的任务，类class路径
MAIN_CLASS="com.flink.job.CommonFlinkStreamJob"
# 我的代码包
JAR_PATH="/home/dev/flink/lib/flink-cdc-1.0.jar"
FLINK_HOME="/home/dev/flink"
FLINK_CHECKPOINT_PATH="/home/dev/flink/flink-checkpoints"
# 配置新得任务时需要修改下方两个配置
# 这个是任务参数，你没有可以不写，取决于你的class任务里面是否设置
JOB_ARGS="--jobKey cdcmysql"
# 这个是你给这个任务的名字
JOB_NAME="cdcmysql"# 获取正在运行的任务id，5秒*10次的容错。
get_job_id() {local retry_count=0                         # 初始化重试计数local max_retries=10                        # 最大重试次数local j_id=""                               # 用于存储Job IDwhile [ $retry_count -lt $max_retries ]; do # 当重试次数小于最大重试次数时# 使用flink命令获取正在运行的任务，并截取到job_idj_id=$($FLINK_HOME/bin/flink list | grep -o '[0-9a-f]\{32\} : '"$JOB_NAME"' (RUNNING)')if [ -n "$j_id" ]; then # 如果成功获取到Job IDbreak                 # 跳出循环fisleep 5 # 等待5秒后重试echo "尝试获取jobid $((retry_count + 1))/$max_retries "retry_count=$((retry_count + 1)) # 重试计数加1doneJ_ID=$j_id # 将获取到的Job ID赋值给全局变量J_ID
}
start_job() {echo "Starting Flink job..."# 检查是否存在之前保存的作业ID文件if [ -f "${FLINK_HOME}/job_id_$JOB_NAME.tmp" ]; then# 从文件中读取最新记录的job_id，空格是必须的job_id=$(grep "${JOB_NAME} " ${FLINK_HOME}/job_id_$JOB_NAME.tmp | cut -d ':' -f 1 | sed 's/ //g')if [ -n "$job_id" ]; then# 查找该job_id的检查点目录checkpoint_dirs=$(find "$FLINK_CHECKPOINT_PATH/$job_id" -type d -name 'chk-*')echo "job[$JOB_NAME]检查点查询，job_id: $job_id, 路径：$checkpoint_dirs"if [ -z "$checkpoint_dirs" ]; then# 路径为空，查找上一个可恢复的检查点路径pre_job_id=$(grep "${JOB_NAME}bak" ${FLINK_HOME}/job_id_$JOB_NAME.tmp | cut -d ':' -f 1 | sed 's/ //g')if [ -n "$pre_job_id" ]; thenpre_checkpoint_dirs=$(find "$FLINK_CHECKPOINT_PATH/$pre_job_id" -type d -name 'chk-*')if [ -z "$pre_checkpoint_dirs" ]; thenecho "没有找到检查点路径，尝试从上一个检查点恢复：${pre_job_id},路径：$pre_checkpoint_dirs"echo "无法恢复任务，停止脚本，请手动设置检查点：从上一次成功执行并且$FLINK_CHECKPOINT_PATH/job_id目录下有chk开头的路径恢复"exit 1elseecho "没有找到检查点路径，尝试从上一个检查点恢复：${pre_job_id},路径：$pre_checkpoint_dirs"job_id=$pre_job_idecho "尝试成功，正在恢复任务进度，从Checkpoint path：$pre_checkpoint_dirs"$FLINK_HOME/bin/flink run -s $pre_checkpoint_dirs -c $MAIN_CLASS $JAR_PATH $JOB_ARGS >>$FLINK_HOME/log/startjob$JOB_NAME.out &fielseecho "没有找到检查点路径，尝试从上一个检查点恢复：${pre_job_id},路径：$pre_checkpoint_dirs"# 没有备份jobidecho "无法恢复任务，停止脚本，请手动设置检查点：从上一次成功执行并且$FLINK_CHECKPOINT_PATH/job_id目录下有chk开头的路径恢复"exit 1fielseecho "正在恢复任务进度，从Checkpoint path：$checkpoint_dirs"$FLINK_HOME/bin/flink run -s $checkpoint_dirs -c $MAIN_CLASS $JAR_PATH $JOB_ARGS >>$FLINK_HOME/log/startjob$JOB_NAME.out &fisleep 10get_job_idif [ -n "$J_ID" ]; thenecho "Job启动脚本执行成功. JobID: $J_ID"echo "$J_ID" >"${FLINK_HOME}/job_id_$JOB_NAME.tmp"# 将上一个可恢复的检查点job作为备份记录echo "${job_id} : ${JOB_NAME}bak" >>"${FLINK_HOME}/job_id_$JOB_NAME.tmp"elseecho "Failed to get JobID after $max_retries retries. Check if the job is started."fifielseecho "job_id_$JOB_NAME.tmp 文件不存在，starting a new job!!!"$FLINK_HOME/bin/flink run -c $MAIN_CLASS $JAR_PATH $JOB_ARGS >>$FLINK_HOME/log/startjob$JOB_NAME.out &sleep 10get_job_idif [ -n "$J_ID" ]; thenecho "Job启动脚本执行成功. JobID: $J_ID"echo "$J_ID" >"${FLINK_HOME}/job_id_$JOB_NAME.tmp"elseecho "Failed to get JobID after $max_retries retries. Check if the job is started."fifi
}
watch_job() {while true; dostatus=$($FLINK_HOME/bin/flink list | grep -o '[0-9a-f]\{32\} : '"$JOB_NAME"' (RUNNING)')if [ -z "$status" ]; thenecho "Flink job is not running. Restarting..."start_jobelseecho "$(date +"%Y-%m-%d %H:%M:%S") Flink job is running."fisleep 180 # 每3分钟检查一次任务状态done
}
watch_job

这个脚本的流程：从tmp文件里面读取jobID如果读取到就查找checkpoints路径从而恢复，如果第一个jobid没有找到tmp文件里面会记录上一次可用检查点的jobid，然后恢复；启动任务后会调用函数有容错的读取jobid从而写入到tmp文件中，并且会使用一个线程一直去监测任务的运行情况，执行上述的逻辑。