Flink 任务启动脚本-V2（包括ck启动）

2. #!/bin/bash
5. #crontab时设置，如果依赖其他环境变量配置，可以在脚本执行一下环境变量脚本
6. source /etc/profile
9. # 进入脚本目录
10. curdir=`dirname "$0"`
11. curdir=`cd "$curdir"; pwd`
12. echo "进入启动脚本目录 $curdir"
16. # 定义应用程序名称
17. APP_NAME="orderTest"
19. # 定义checkpoint路径
20. CHECKPOINT_BASE_PATH="hdfs:///jobs/flink/checkpoints/$APP_NAME/"
22. MAIN_CLASS="com.test.mainTest"
24. #绝对路径
25. JAR_PATH="/$curdir/flink-test-1.0.0.jar"
27. # 默认从checkpoint启动
28. USE_CHECKPOINT=true
31. # 显示使用说明
32. usage() {
33.     echo "用法: $0 [-n] [-h]"
34.     echo "选项："
35.     echo "  -n    不从checkpoint启动任务（默认从最新的checkpoint启动）"
36.     echo "  -h    显示此帮助信息"
37.     echo
38.     echo "示例："
39.     echo "  $0        # 从最新的checkpoint启动任务"
40.     echo "  $0 -n     # 不使用checkpoint启动任务"
41.     exit 1
42. }
44. # 解析命令行参数
45. while getopts ":nh" opt; do
46.     case $opt in
47.         n)
48.             USE_CHECKPOINT=false
49.             echo "已设置：不从checkpoint启动任务"
50.             ;;
51.         h)
52.             usage
53.             ;;
54.         \?)
55.             echo "错误：无效的选项 -$OPTARG"
56.             echo "使用 -h 查看帮助信息"
57.             exit 1
58.             ;;
59.     esac
60. done
62. # 如果设置了无效参数，显示使用说明
63. if [ $OPTIND -gt 1 ]; then
64.     shift $((OPTIND-1))
65.     if [ "$#" -gt 0 ]; then
66.         echo "错误：存在额外的参数 $@"
67.         echo "使用 -h 查看帮助信息"
68.         exit 1
69.     fi
70. fi
73. # 检查是否存在指定应用程序在运行中
74. is_running=$(yarn application -list | grep -w "$APP_NAME" | grep -c "RUNNING")
76. if [ $is_running -gt 0 ]; then
77.     echo "应用程序 $APP_NAME 在运行中，退出脚本"
78.     exit 1
79. else
80.     echo "应用程序 $APP_NAME 不在运行中，准备拉起任务"
81. fi
83. # 函数：获取最新成功的checkpoint地址
84. get_latest_checkpoint() {
85.     latest_checkpoint=$(hdfs dfs -ls -t -R $CHECKPOINT_BASE_PATH  | grep '_metadata' | sort -k6,7r | head -n 1 | awk '{print $8}'  )
86.     if [ -z "$latest_checkpoint" ]; then
87.         return 1
88.     fi
89.     # 去掉文件名，只保留目录路径
90.     checkpoint_dir=$(dirname "$latest_checkpoint")
91.     echo $checkpoint_dir
92.     return 0
93. }
95. # 构建基础命令
96. CMD="flink run \
97. -t yarn-per-job \
98. -d \
99. -p 1 \
100. -Dyarn.application.queue=realtime \
101. -Dyarn.application.name=$APP_NAME \
102. -Dtaskmanager.memory.process.size=4096mb \
103. -Dtaskmanager.memory.network.max=64mb \
104. -Dtaskmanager.memory.managed.size=1024mb \
105. -Dtaskmanager.numberOfTaskSlots=1 \
106. -c $MAIN_CLASS"
108. if [ "$USE_CHECKPOINT" = true ]; then
109.     # 获取最新的checkpoint地址
110.     LATEST_CHECKPOINT=$(get_latest_checkpoint)
111.     if [ $? -ne 0 ]; then
112.         echo "没有找到适合的ck，退出执行"
113.         exit 1
114.     fi
115.     echo "'$APP_NAME' 任务将从 '$LATEST_CHECKPOINT' 启动"
116.     CMD="$CMD \
117.     -s $LATEST_CHECKPOINT"
118. else
119.     echo "'$APP_NAME' 任务将不从checkpoint启动"
120. fi
122. # 添加最终的jar包和配置文件参数
123. CMD="$CMD \
124.    $JAR_PATH "
126. # 执行命令
127. echo " 任务启动命令: '$CMD' "
129. eval $CMD

复制代码

本次新增一个不从 ck 启动的选择，默认从 最新 ck 启动，用法 sh start.sh -n 。
备注：由于Flink checkpoint 个别情况下，不一定能保证落地的checkpoint文件一定有效，所以需要人工介入支持。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

继续阅读请点击广告