kill_idle_spark_app_on_yarn.py

import sys, requests, json, time, datetime
from subprocess import call


def getLastEndJob(url, startTime):
  resp = requests.get(url + "/api/v1/applications")
  resp.encoding = 'utf-8'
  sparkuis = resp.json()
  kill = True
  msg = ""
  lastJob = startTime
  for sparkui in sparkuis:
    name = str(sparkui['name'])
    if (name == "Thrift JDBC/ODBC Server"):
      msg = "Spark Thrift Server will not be killed"
      kill = False
    resp = requests.get(url + "/api/v1/applications/" + str(sparkui['id']) + "/jobs/")
    resp.encoding = 'utf-8'
    jobs = resp.json()
    for job in jobs:
      status = str(job['status'])
      if (status == "RUNNING"):
        msg = "Spark job running, will not be killed"
        kill = False
        lastJob = datetime.datetime.now().strftime("%Y-%m-%dT%H:%M:%S") + ".000GMT"
      else:
        submission = str(job['submissionTime'])
        #print("\t\t\t" + submission)
        if (submission > lastJob):
          lastJob = submission
  #print("\t\tlastJob=" + lastJob)
  idle = datetime.datetime.now() - datetime.datetime.strptime(lastJob[:-7], "%Y-%m-%dT%H:%M:%S")
  if (kill):
    if (idle.total_seconds() > (60*60)):
      kill = True
      msg = "idle for more than 1 hour"
    else:
      kill = False
      msg = "idle for less than 1 hour"
  return (kill, msg, idle)


rmServer = "ip-172-31-13-63"
rmPort = "8088"
url = "http://" + rmServer + ":" + rmPort + "/ws/v1/cluster/apps?states=running"

resp = requests.get(url)
resp.encoding = 'utf-8'
applications = resp.json()


print("*******************************************")
print("SPARK APP KILLER - STARTED AT " + datetime.datetime.now().strftime("%Y-%m-%dT%H:%M:%S"))
print("-----------------")
print("STATUS BEFORE KILLING APPLICATION")
print("-----------------")

print("ID\tNAME\tUSERNAME\tQUEUE\t# CONTAINERS\tGOING TO BE KILLED\tMSG\tLAST JOB\tDELTA LAST JOB\tSECONDS DELTA")
for application in applications['apps']['app']:
  startTime = datetime.datetime.fromtimestamp(int(str(application['startedTime']))/1000).strftime("%Y-%m-%dT%H:%M:%S") + ".000GMT"
  (kill, msg, idle) = getLastEndJob(application['trackingUrl'],startTime)
  print(str(application['id']) + "\t"  + str(application['name']) + "\t" + str(application['user']) + "\t" + str(application['queue']) + "\t" + str(application['runningContainers']) + "\t" + str(kill) + "\t" + msg + "\t" + str(idle) + "\t" + str(idle.total_seconds()))


print("-----------------")
print("START KILLING APPS")
print("-----------------")
for application in applications['apps']['app']:
  startTime = datetime.datetime.fromtimestamp(int(str(application['startedTime']))/1000).strftime("%Y-%m-%dT%H:%M:%S") + ".000GMT"
  (kill, msg, idle) = getLastEndJob(application['trackingUrl'],startTime)
  if (kill):
    print("Killing Application: " + str(application['id']) + "\t"  + str(application['name']) + "\t" + str(application['user']) + "\t" + str(application['queue']) + "\t" + str(application['runningContainers']) + "\t" + str(kill) + "\t" + msg + "\t" + str(idle) + "\t" + str(idle.total_seconds()))
    call(["yarn", "application", "--kill", str(application['id'])])
    print("Application Killed: " + str(application['id']))


print("-----------------")
print("STATUS AFTER KILLING APPLICATION")
print("-----------------")

resp = requests.get(url)
resp.encoding = 'utf-8'
applications = resp.json()


print("ID\tNAME\tUSERNAME\tQUEUE\t# CONTAINERS\tGOING TO BE KILLED\tMSG\tLAST JOB\tDELTA LAST JOB\tSECONDS DELTA")
for application in applications['apps']['app']:
  startTime = datetime.datetime.fromtimestamp(int(str(application['startedTime']))/1000).strftime("%Y-%m-%dT%H:%M:%S") + ".000GMT"
  (kill, msg, idle) = getLastEndJob(application['trackingUrl'],startTime)
  print(str(application['id']) + "\t"  + str(application['name']) + "\t" + str(application['user']) + "\t" + str(application['queue']) + "\t" + str(application['runningContainers']) + "\t" + str(kill) + "\t" + msg + "\t" + str(idle) + "\t" + str(idle.total_seconds()))


print("-----------------")

print("SPARK APP KILLER - FNISHED AT " + datetime.datetime.now().strftime("%Y-%m-%dT%H:%M:%S"))


print("*******************************************")