企业门户网站开发源码宜兴百度推广公司
Flink任务一般为实时不断运行的任务，如果没有任务监控,
任务异常时无法第一时间处理会比较麻烦。
这里通过调用API接口方式来获取参数，实现任务监控。
Flink任务监控（基于API接口编写shell脚本）
一 flink-on-yarn 模式
二 编写shell 脚本
监控集群指标
http：//rm-http-address：port/ws/v1/cluster/metrics
响应正文
<clusterMetrics>
<appsSubmitted>**</appsSubmitted>
<appsCompleted>**</appsCompleted>
<appsPending>0</appsPending>
<appsRunning>**</appsRunning>
<appsFailed>**</appsFailed>
<appsKilled>**</appsKilled>
<reservedMB>0</reservedMB>
<availableMB>**</availableMB>
<allocatedMB>**</allocatedMB>
<pendingMB>0</pendingMB>
<reservedVirtualCores>0</reservedVirtualCores>
<availableVirtualCores>**</availableVirtualCores>
<allocatedVirtualCores>**</allocatedVirtualCores>
<pendingVirtualCores>0</pendingVirtualCores>
<containersAllocated>**</containersAllocated>
<containersReserved>0</containersReserved>
<containersPending>0</containersPending>
<totalMB>**</totalMB>
<totalVirtualCores>**</totalVirtualCores>
<utilizedMBPercent>53</utilizedMBPercent>
<utilizedVirtualCoresPercent>**</utilizedVirtualCoresPercent>
<rmSchedulerBusyPercent>0</rmSchedulerBusyPercent>
<totalNodes>**</totalNodes>
<lostNodes>0</lostNodes>
<unhealthyNodes>**</unhealthyNodes>
<decommissioningNodes>0</decommissioningNodes>
<decommissionedNodes>0</decommissionedNodes>
<rebootedNodes>0</rebootedNodes>
<activeNodes>**</activeNodes>
<shutdownNodes>0</shutdownNodes>
<totalAllocatedContainersAcrossPartition>0</totalAllocatedContainersAcrossPartition>
<crossPartitionMetricsAvailable>false</crossPartitionMetricsAvailable>
</clusterMetrics>
clusterMetrics 对象的元素
| 项目 | 数据类型 | 描述 |
|---|---|---|
| apps已提交 | int | 提交的申请数量 |
| 应用已完成 | int | 完成的申请数量 |
| apps待定 | int | 待处理的申请数量 |
| 应用程序正在运行 | int | 正在运行的应用程序数 |
| apps失败 | int | 失败的应用程序数 |
| 应用已杀死 | int | 被终止的应用程序数 |
| 保留MB | 长 | 预留的内存量（以 MB 为单位） |
| 可用MB | 长 | 可用内存量（以 MB 为单位） |
| 已分配MB | 长 | 分配的内存量（以 MB 为单位） |
| 总MB | 长 | 总内存量（以 MB 为单位） |
| 保留虚拟核心 | 长 | 保留的虚拟核心数 |
| 可用虚拟核心 | 长 | 可用虚拟核心数 |
| 分配的虚拟核心 | 长 | 分配的虚拟核心数 |
| totalVirtualCores 虚拟核心数 | 长 | 虚拟核心总数 |
| 容器已分配 | int | 分配的容器数 |
| 容器保留 | int | 保留的容器数 |
| 容器挂起 | int | 待处理的容器数 |
| 总节点数 | int | 节点总数 |
| 活动节点 | int | 活动节点数 |
| 丢失节点 | int | 丢失的节点数 |
| 不健康的节点 | int | 不正常的节点数 |
| 停用节点 | int | 停用的节点数 |
| 已停用节点 | int | 停用的节点数 |
| rebooted节点 | int | 重新启动的节点数 |
| shutdown节点 | int | 关闭的节点数 |
获取所有application
curl -s http://XXX:8088/ws/v1/cluster/apps
获取 state值为 RUNNING 的application任务
curl -s http://XXX:8088/ws/v1/cluster/apps?state=RUNNING
获取这个任务单个信息
curl -s http://XXX:8088/ws/v1/cluster/apps/application_1619074605427_0063 |jq .app.state
请注意，根据安全设置，用户可能无法看到所有字段。
项目 数据类型 描述 编号 字符串 应用程序 ID 用户 字符串 启动应用程序的用户 名字 字符串 应用程序名称 队列 字符串 提交应用程序的队列 州 字符串 根据 ResourceManager 的应用程序状态 - 有效值是 YarnApplicationState 枚举的成员：NEW、NEW_SAVING、SUBMITTED、ACCEPTED、RUNNING、FINISHED、FAILED、KILLED finalStatus 字符串 应用程序的最终状态（如果已完成）（由应用程序本身报告）有效值是 FinalApplicationStatus 枚举的成员：UNDEFINED、SUCCEEDED、FAILED、KILLED 进展 浮 以百分比表示的申请进度 trackingUI 字符串 跟踪 URL 当前指向的位置 - 历史记录（用于历史记录服务器）或 ApplicationMaster trackingUrl 字符串 可用于跟踪应用程序的 Web URL 诊断 字符串 详细的诊断信息 clusterId 长 集群 ID 应用程序类型 字符串 应用程序类型 application标签 字符串 应用程序的逗号分隔标记 优先权 字符串 所提交申请的优先权 开始时间 长 应用程序启动的时间（自纪元以来的毫秒） 完成时间 长 应用程序完成的时间（以纪元以来的毫秒数为单位） elapsedTime 长 自应用程序启动以来经过的时间（以毫秒为单位） amContainer日志 字符串 应用程序主容器日志的 URL amHostHttp地址 字符串 应用程序主机的节点 http 地址 amRPCAddress 字符串 应用程序主机的 RPC 地址 已分配MB int 分配给应用程序正在运行的容器的内存总和（以 MB 为单位） 已分配VCores int 分配给应用程序正在运行的容器的虚拟核心的总和 running容器 int 当前为应用程序运行的容器数 memorySeconds 长 应用程序分配的内存量（兆字节-秒） vcore秒数 长 应用程序分配的 CPU 资源量（虚拟内核 - 秒） queueUsagePercentage 浮 应用正在使用的队列资源的百分比 clusterUsage百分比 浮 应用正在使用的群集资源的百分比。 抢占ResourceMB 长 抢占式容器使用的内存 preemptedResourceVCores 长 抢占容器使用的虚拟核心数 numNonAMContainer抢占 int 抢占的标准容器数 numAMContainer抢占 int 抢占的应用程序主容器数 logAggregationStatus 字符串 日志聚合的状态 - 有效值是 LogAggregationStatus 枚举的成员：DISABLED、NOT_START、RUNNING、RUNNING_WITH_FAILURE、SUCCEEDED、FAILED、TIME_OUT unmanaged应用程序 布尔 应用程序是否处于非托管状态。 appNodeLabelExpression 字符串 节点标签表达式，用于标识默认情况下应在其上运行应用程序容器的节点。 amNodeLabel表达式 字符串 节点标签表达式，用于标识应用程序的 AM 容器预期在其上运行的节点。
jq，是linux一个很方便的json处理工具
通俗的说就是一个能够接受json，处理json，输出json的程序，反正很好用。
安装起来也非常的方便，直接使用yum即可安装。linux下离线安装jq工具 - 代码天地 (codetd.com)
yum install jq
编写shell脚本
由于公司离线yarn和实时yarn 采用是分开的方式。
只需要监控实时yarn 任务有没有处于RUNNING，达到监控的目的
这里shell脚本也只记录，flink-on-yarn 这种部署方式任务监控
shell脚本水平有限，大家多多谅解，欢迎指导shell脚本实现功能：
获取线运行job任务，记录到日志文件。下一次脚本调用时候读取日志文件，判断状态。
不是RUNNING，就告警同时重新记录日志。
#!/bin/bashJoblist=`cat /opt/shell/logs/flink_job.log` #获取记录job的log文件
let i=0 #获取任务数
let log_count=0 #获取日志中的任务数
start_count=RUNNING #判断任务是否存在异常############## 1 判断日志文件内容是否为空，为空时自动读取flink任务并记录到日志文件 #########
if [ -z "$Joblist" ]
thenwhile :dojob_id[$i]=`curl -s http://XXX:8088/ws/v1/cluster/apps?state=RUNNING |jq .apps.app[$i].id`if [ ${job_id[$i]} = "null" ];thenbreakelseecho ${job_id[$i]}echo ${job_id[$i]}>>/opt/shell/logs/flink_job.loglet i++fidone
fi############## 2 读取文件中JOB任务 ##################let i=0
while read line
doJOB[$i]=$linelet i++
done</opt/shell/logs/flink_job.loglog_count=$i #获取日志中的任务数########### 3 判断任务状态，是否为RUNNIG，不是则邮件告警 ###############
for ((j=0;j<i;j++))
doJOB_ID=${JOB[$j]//\"}JOB_status=`curl -s http://XXXX:8088/ws/v1/cluster/apps/$JOB_ID | jq .app.state`JOB_NAME=`curl -s http://XXX:8088/ws/v1/cluster/apps/$JOB_ID | jq .app.name`START=$[`curl -s http://XXX:8088/ws/v1/cluster/apps/$JOB_ID | jq .app.startedTime` / 1000]# echo "JOB_NAME: "$JOB_NAME
# echo 启动时间: `date -d @$START +"%F %H:%M:%S"`
# echo "JOB_status: " ${JOB_status//\"}#echo -e "【$JOB_NAME】 \n JOB_ID: $JOB_ID \n 启动时间: `date -d @$START +"%F %H:%M:%S"` \n 检查时间: `date "+%Y-%m-%d %H:%M:%S"` \n 目前状态: $JOB_status"
#echo "=============================================="if [ ${JOB_status//\"} != "RUNNING" ];thenSUBJECT="【异常告警】Flink任务异常"TEXT="Flink任务 【$JOB_NAME】 异常故障 \n\nJOB_ID: $JOB_ID\n\n启动时间: `date -d @$START +"%F %H:%M:%S"` \n\n检查时间: `date "+%Y-%m-%d %H:%M:%S"` \n\n目前状态: $JOB_status"echo -e $TEXT | mail -s $SUBJECT 邮箱地址start_count=erronfi
done########### 4 出现任务异常，重新读取job 任务记录到日志文件 ###############let i=0
if [ $start_count == "erron" ];thenecho '重新写入日志文件'while :dojob_id[$i]=`curl -s http://XXXX:8088/ws/v1/cluster/apps?state=RUNNING |jq .apps.app[$i].id`if [ ${job_id[$i]} = "null" ];thenbreakelif [ $i == 0 ]; thenecho ${job_id[$i]}>/opt/shell/logs/flink_job.logelseecho ${job_id[$i]}>>/opt/shell/logs/flink_job.logfilet i++donestart_count=RUNNING
fi########### 5 判断线上任务数是否一致,是否有新任务增加 ###############let i=0
while :
dojob_id[$i]=`curl -s http://XXX:8088/ws/v1/cluster/apps?state=RUNNING |jq .apps.app[$i].id`if [ ${job_id[$i]} = "null" ];thenbreakelselet i++fi
done
let count=$i #线上任务数
echo "==========================线上最新RUNNING状态任务数: "$count
echo "==========================日志RUNNING状态任务数: "$log_countif [ ! $count -eq $log_count ]; thenecho "现有RUNNING状态任务数不相等于已记录的任务数"echo ${job_id[0]} >/opt/shell/logs/flink_job.logfor ((i=1;i<count;i++))doecho "重新写入JOB: "${job_id[$i]}echo ${job_id[$i]}>> /opt/shell/logs/flink_job.logdonefiecho "======================当前时间: `date "+%Y-%m-%d %H:%M:%S"`======================================="
echo ================================================================================================
echo =====================================本次crontab监控结束========================================
echo ================================================================================================Yarn REST API 使用指南-阿里云开发者社区
Apache Hadoop 3.0.1 – ResourceManager REST API。