任务重启

重启脚本

#!/usr/bin/env python3
import requests
import pymysql
import datetime
import subprocess

# YARN集群的地址
yarn_url = "http://dmcloud17:8088"
yarn_job_list = []

# YARN任务类
class Job:
    def __init__(self, name, app_id, state):
        self.name = name
        self.app_id = app_id
        self.state = state


api_url = f"{yarn_url}/ws/v1/cluster/apps?state=RUNNING"

try:
    response = requests.get(api_url)

    if (response.status_code == 200) and (response.json()['apps'] is not None):
        yarn_list = response.json()['apps']['app']
        for yarn_job in yarn_list:
            job = Job(str(yarn_job['name']), str(yarn_job['id']), str(yarn_job['state']))
            yarn_job_list.append(job)
    else:
        print("无法连接到YARN集群或未找到正在运行的任务.")
except Exception as e:
    print("发生错误:", str(e))


conn = pymysql.connect(host='ip', user='streampark', password='streampark', database='streampark')
cursor = conn.cursor()


class MySQLJob:
    def __init__(self, id, job_name, app_id, state):
        self.id = id
        self.job_name = job_name
        self.app_id = app_id
        self.state = state


mysql_job_list = []
try:
    sql = """
        select id, job_name, app_id, state from t_flink_app
    """
    cursor.execute(sql)
    result = cursor.fetchall()
    for row in result:
        mysql_job = MySQLJob(str(row[0]), str(row[1]), str(row[2]), str(row[3]))
        mysql_job_list.append(mysql_job)
except Exception as e:
    print("Error: ", str(e))
finally:
    cursor.close()
    conn.close()


if len(mysql_job_list) > 0:

    need_kill_job_list = []
    for mysql_job in mysql_job_list:
        for yarn_job in yarn_job_list:
            if mysql_job.app_id != yarn_job.app_id and mysql_job.job_name == yarn_job.name:
                need_kill_job_list.append(yarn_job)


    for kill_job in need_kill_job_list:
        command = f'yarn application -kill {kill_job.app_id}'
        print('failed job id: ' + kill_job.app_id + ', job_name: ' + kill_job.name + ', state: ' + kill_job.state + ', time: ' + str(datetime.datetime.now()))
        subprocess.run(command, shell=True)

    # 2. 第一次登录，获取token
    url_sign = "http://bigdata:10000/passport/signin"
    data_sign = {"username": "admin", "password": "streampark", "loginType": "PASSWORD"}
    authorization = ""
    response = requests.post(url_sign, data=data_sign)
    if response.status_code == 200:
        authorization = response.json()['data']['token']

    # 3. 使用token发送启动任务
    url_job = "http://bigdata:10000/flink/app/start"
    headers = {"Authorization": authorization}
    for mysql_job in mysql_job_list:
        if mysql_job.state in ('7', '9'):
            data = {"id": mysql_job.id, "savePointed": "false", "savePoint": "", "allowNonRestored": "false", "teamId": 100000}
            r = requests.post(url=url_job, data=data, headers=headers)
            if r.status_code == 200:
                print(mysql_job.job_name + 'has been started!!!' + str(datetime.datetime.now()))