没有监控的流处理作业与茫茫大海中的裸泳无异 - 附 flink 与 spark 作业监控脚本实现

#!/usr/bin/env bash`
#overviewUrl="http://10.20.39.42:8081/overview"
#jobsOverviewUrl="http://10.20.39.42:8081/jobs/overview"
#expectedNumOfTm=2
#jobName="Socket Window WordCount"
overviewUrl=$1
jobsOverviewUrl=$2
expectedNumOfTm=$3
jobName=$4
#check flink job manager state and exit if not good
overviewResult=`curl -s -X GET "$overviewUrl"`
if [ -z "$overviewResult" ]
then
 echo "flink job manager is not running properly, exit!"
 exit 1
else
 echo "flink job manager is in good health."
fi
#check flink task manager state and exit if not good
numOfTm=`echo $overviewResult | jq .taskmanagers`
if [ "$numOfTm" -ne "$expectedNumOfTm" ]; 
then 
 echo "number of living task manager is not as expected, exit!"
 exit 2
else 
 echo "flink task manager is in good state."
fi
#check flink job state and exit if not good
jobsOverviewResult=`curl -s -X GET "$jobsOverviewUrl"`
jobState=`echo $jobsOverviewResult | jq --arg jobName "$jobName" -r '.jobs[] | select(.name == $jobName and .state == "RUNNING")'`
if [ -z "$jobState" ]
then
 echo "there are no jobs with name $jobName in running state, exit!"
 exit 3
else
 echo "flink job $jobName is in running state."
fi

#!/bin/bash
# specify input paramaters here, which are yarn resourcemanger address, and the spark streaming application name you want to monitor
#params=(192.168.71.69:8088 spark-streaming-app-name)
params=("$@")
yarnRm=${params[0]}
appName=${params[1]}
#echo "the input yarn rm is；" $yarnRm
#echo "the input appName to be monitored is: " $appName
#get application id and tracking url from yarn resource manager
yarnRestUrl="http://$yarnRm/ws/v1/cluster/apps?states=RUNNING&applicationTypes=spark"
yarnAppDetails=`curl -s "$yarnRestUrl"`
#echo "yarnRestUrl is: " $yarnRestUrl
#echo "yarnAppDetails is: " $yarnAppDetails
# if no spark application is in running state, exit with code 1
if [ -z "$yarnAppDetails" ]
then
 #echo "no spark application is in running status, exit!"
 echo "2"
 exit 1
fi
yarnAppId=`echo $yarnAppDetails | jq --arg appName "$appName" -r '.apps | select(.app[].name == $appName) | .app[0].id'`
yarnAppTrackingUrl=`echo $yarnAppDetails | jq --arg appName "$appName" -r '.apps | select(.app[].name == $appName) | .app[0].trackingUrl'`
#echo "yarnAppId is: " $yarnAppId
#echo "yarn application tracking url is: " $yarnAppTrackingUrl
#if no spark application with specified name is in running state, exit
if [ -z "$yarnAppId" ]
then
 #echo "no spark application with name $appName is in running status, exit!"
 echo "2"
 exit 1
fi
#get application completed batches from spark
sparkRestUrl="$yarnAppTrackingUrl/api/v1/applications/$yarnAppId/streaming/statistics"
sparkAppDetails=`curl -s "$sparkRestUrl"`
sparkBatchDuration=`echo $sparkAppDetails | jq .batchDuration`
sparkCompletedBatchesPrev=`echo $sparkAppDetails | jq .numTotalCompletedBatches`
sparkCheckInterval=$((sparkBatchDuration/1000*2))
#echo "spark restful url is: " $sparkRestUrl
#echo $sparkAppDetails
#echo "spark batch duration is: " $sparkBatchDuration
#echo $sparkCompletedBatchesPrev
#echo $sparkCheckInterval
 sleep $sparkCheckInterval
 sparkAppDetails=`curl -s "$sparkRestUrl"`
 sparkCompletedBatchesCurrent=`echo $sparkAppDetails | jq .numTotalCompletedBatches`
 if [ "$sparkCompletedBatchesCurrent" -le "$sparkCompletedBatchesPrev" ]
 then
     # spark streaming 程序在2个 batch duration 内1个 batch 都没有完成，处于不健康状态，退出！
     #echo "spark streaming application $appName is in bad health!!! sparkCompletedBatchesPrev is: $sparkCompletedBatchesPrev, sparkCompletedBatchesCurrent is $sparkCompletedBatchesCurrent"
     echo "1"
     exit 1 
 else
     #echo "spark streaming application $appName is in good health!!! sparkCompletedBatchesPrev is: $sparkCompletedBatchesPrev, sparkCompletedBatchesCurrent is $sparkCompletedBatchesCurrent"
     echo "0"
 fi