Linux命令行：

sh flink_run.sh test com.dzj.app.base.test.FlinkKafkaOffsetTest /root/soft/test.jar

flink_run.sh脚本内容：

#!/bin/bash# 检查参数数量是否正确
if [ "$#" -ne 3 ]; thenecho "错误：需要提供 3 个参数"exit 1
else# 打印所有输入参数echo "输入参数数量：$#"echo "输入参数列表：$@"
fi# 打印每个参数
echo "第一个参数 -ynm：$1"
echo "第二个参数 -c：$2"
echo "第三个参数 jar：$3"# 获取所有正在运行的 YARN 任务
running_tasks=$(yarn application -list | grep RUNNING)# 检查是否有名为 $1 的任务
if echo "$running_tasks" | grep -q "$1"; then# 如果找到了名为 $1 的任务echo "Yarn-Cluster此运行任务已在运行：Found '$1' task running"exit 0
else# 如果没有找到名为 $1 的任务，返回 0echo "Yarn-Cluster未找到此运行任务：No '$1' task running"# 获取 /flink/checkpoint/$1/ 目录下最新修改的文件夹latest_folder=$(hdfs dfs -ls /flink/checkpoint/$1/ | grep '^d' | awk '{print $6, $7, $8}' | sort -k1,1 -k2,2 | tail -1 | awk '{print $3}')latest_ckpt_folder=$(hdfs dfs -ls $latest_folder | grep '^d' | awk '{print $6, $7, $8}' | sort -k1,1 -k2,2 | tail -1 | awk '{print $3}')# 检查是否找到最新文件夹的路径if [ -z "$latest_ckpt_folder" ]; thenecho "未找到最新的checkpoint路径No ckpt_folder in： /flink/checkpoint/$1/"# 替换命令中的 -s 参数为最新文件夹的路径flink_command="flink run -d -m yarn-cluster -ynm $1 -c $2 $3"else# 返回最新修改的文件夹名echo "最新的checkpoint路径Latest ckpt_folder: $latest_ckpt_folder"# 替换命令中的 -s 参数为最新文件夹的路径flink_command="flink run -s hdfs://172.29.105.99:8020$latest_ckpt_folder -m yarn-cluster -ynm $1 -c $2 $3"fi
fi# 执行命令
echo "Flink执行命令Executing command: $flink_command"
eval $flink_command