from asyncio import current_task import json import time from turtle import update from app import crud, models, schemas from app.common import minio from app.core.datasource.datasource import DataSourceBase from app.crud.jm_homework_datasource_relation import get_jm_relations from app.utils.send_util import * from sqlalchemy.orm import Session from app.common.minio import minio_client from configs.settings import DefaultOption, config DATABASE_NAME = config.get('HIVE', 'DATABASE_NAME') requirement_path = config.get('REQUIREMENTS_CONFIG', 'path') requirement_prefix = config.get('REQUIREMENTS_CONFIG', 'prefix') type_dict = { "Java": "java", "Python": "python", "Dag": "sparks" } def jm_job_create_task(jm_homework: models.JmHomework, relation_list, db: Session): content = '' envs = {} if jm_homework.type == "Dag": content = red_dag_and_format(jm_homework, relation_list, db) requirements_relation = crud.get_requirements_relation(db, jm_homework.dag_uuid) if requirements_relation: envs.update({'requirement_package_path': f'{requirement_prefix}{requirement_path}/dag_{jm_homework.dag_uuid.lower()}.zip'}) elif jm_homework.type == "Python": content = red_python_and_format(jm_homework) af_task = { "name": jm_homework.name, "file_urls": [] if jm_homework.type != "Java" else [jm_homework.script_file], "script": content if jm_homework.type != "Java" else "", "cmd": jm_homework.execute_command if jm_homework.type != "Dag" else "", "cmd_parameters": "", "envs": envs, "run_image": jm_homework.image_url if jm_homework.type != "Dag" else "", "task_type": type_dict[jm_homework.type], "user_id": 0, } res = send_post('/af/af_task', af_task) af_task = res['data'] return af_task def jm_job_update_task(jm_homework: models.JmHomework, relation_list, db: Session): relation = crud.get_af_id(db, jm_homework.id, 'task') content = '' envs = {} if jm_homework.type == "Dag": content = red_dag_and_format(jm_homework, relation_list, db) requirements_relation = crud.get_requirements_relation(db, jm_homework.dag_uuid) if requirements_relation: envs.update({'requirement_package_path': f'{requirement_prefix}{requirement_path}/dag_{jm_homework.dag_uuid.lower()}.zip'}) elif jm_homework.type == "Python": content = red_python_and_format(jm_homework) af_task = { "name": jm_homework.name, "file_urls": [] if jm_homework.type != "Java" else [jm_homework.script_file], "script": content if jm_homework.type != "Java" else "", "cmd": jm_homework.execute_command if jm_homework.type != "Dag" else "", "cmd_parameters": "", "envs": envs, "run_image": jm_homework.image_url if jm_homework.type != "Dag" else "", "task_type": type_dict[jm_homework.type], "user_id": 0, } res = send_put('/af/af_task', relation.af_id, af_task) af_task = res['data'] return af_task def jm_job_create_job(jm_job_info: models.JmJobInfo, nodes, edges, db: Session): homework_ids = [node['homework_id'] for node in nodes] node_uuid_to_h_id = {node['id']:node['homework_id'] for node in nodes} relations = crud.get_af_ids(db,homework_ids, 'task') se_id_to_af_id_dict = { relation.se_id:relation.af_id for relation in relations} tasks = [ send_get("/af/af_task/getOnce",id)['data'] for id in se_id_to_af_id_dict.values()] dependence = [[se_id_to_af_id_dict[node_uuid_to_h_id[str(edge['source'])]],se_id_to_af_id_dict[node_uuid_to_h_id[str(edge['target'])]]] for edge in edges] cron = jm_job_info.cron_expression if jm_job_info.cron_type == 2 else '@once' cron = cron.replace('?','*') af_job = { "tasks": tasks, "name": jm_job_info.name, "dependence": dependence, "cron": cron, "desc": jm_job_info.name, "route_strategy": "", "block_strategy": "", "executor_timeout": 0, "executor_fail_retry_count": 0, "trigger_status": jm_job_info.status, "job_mode":1, "job_type": 0, "user_id": 0, } res = send_post('/af/af_job', af_job) af_job = res['data'] send_submit(af_job['id']) return af_job def jm_job_update_job(jm_job_info: models.JmJobInfo, nodes, edges, db: Session): homework_ids = [node['homework_id'] for node in nodes] node_uuid_to_h_id = {node['id']:node['homework_id'] for node in nodes} relations = crud.get_af_ids(db,homework_ids, 'task') se_id_to_af_id_dict = { relation.se_id:relation.af_id for relation in relations} tasks = [ send_get("/af/af_task/getOnce",id)['data'] for id in se_id_to_af_id_dict.values()] dependence = [[se_id_to_af_id_dict[node_uuid_to_h_id[str(edge['source'])]],se_id_to_af_id_dict[node_uuid_to_h_id[str(edge['target'])]]] for edge in edges] cron = jm_job_info.cron_expression if jm_job_info.cron_type == 2 else '@once' cron = cron.replace('?','*') af_job = { "tasks": tasks, "name": jm_job_info.name, "dependence": dependence, "cron": cron, "desc": jm_job_info.name, "route_strategy": "", "block_strategy": "", "executor_timeout": 0, "executor_fail_retry_count": 0, "trigger_status": jm_job_info.status, } job_relation = crud.get_af_id(db,jm_job_info.id,'job') res = send_put('/af/af_job', job_relation.af_id, af_job) af_job = res['data'] send_submit(af_job['id']) def red_dag_and_format(jm_homework: models.JmHomework, relation_list, db: Session): node_relation_dict = { relation['node_uuid']:relation for relation in relation_list} file = minio_client.get_file(jm_homework.dag_url) result = json.loads(file) edges = result['edges'] # t_s = {} # input_num = {} # for edge in edges: # if edge['target'] in t_s.keys(): # t_s[edge['target']].append(edge['source']) # else: # t_s.update({edge['target']:[edge['source']]}) nodes = result['nodes'] sub_nodes = [] for node in nodes: if node['op'] == 'datasource': fileds = node['data']['input_source'] script = 'select ' for filed in fileds: script += filed['dataField'] + ',' script = script.strip(',') ds_id = node_relation_dict[node['id']]['datasource_id'] database_name = "" if ds_id == -1: database_name = DATABASE_NAME else: data_source = crud.get_job_jdbc_datasource(db,ds_id) database_name = data_source.database_name script += ' from `' + database_name + '`.`'+node_relation_dict[node['id']]['table']+ '`' sub_node = { "id": node['id'], "name": node['name'], "op": 'sql', "script":script } sub_nodes.append(sub_node) elif node['op'] == 'outputsource': fileds = node['data']['output_source'] ds_id = node_relation_dict[node['id']]['datasource_id'] database_name = "" if ds_id == -1: database_name = DATABASE_NAME else: data_source = crud.get_job_jdbc_datasource(db,ds_id) database_name = data_source.database_name script = '''def main_func (input0, spark,sc): input0.write.mode("overwrite").saveAsTable("''' + database_name + '.'+node_relation_dict[node['id']]['table']+'''")''' inputs = node['data']['inputs'] if 'inputs' in node['data'].keys() else {} # index = 0 # input_list = t_s[node['id']] if node['id'] in t_s.keys() else [] # for input in input_list: # if input in input_num.keys(): # input_num[input]+=1 # else: # input_num.update({input:0}) # inputs.update({'input'+str(index):[input,input_num[input]]}) # index+=1 sub_node = { "id": node['id'], "name": node['name'], "op": 'pyspark', "inputs": inputs, "script":script } sub_nodes.append(sub_node) else: inputs = node['data']['inputs'] if 'inputs' in node['data'].keys() else {} # index = 0 # input_list = t_s[node['id']] if node['id'] in t_s.keys() else [] # for input in input_list: # if input in input_num.keys(): # input_num[input]+=1 # else: # input_num.update({input:0}) # inputs.update({'input'+str(index):[input,input_num[input]]}) # index+=1 script = node['data']['script'] if node['op'] == 'sql': script = script.replace('\n', ' ') sub_node = { "id": node['id'], "name": node['name'], "op": node['op'], "inputs": inputs, "script": script, } sub_nodes.append(sub_node) res = { 'sub_nodes': sub_nodes, 'edges': [(edge['source'],edge['target']) for edge in edges] } return json.dumps(res) def red_python_and_format(jm_homework): file = minio_client.get_file(jm_homework.script_file if jm_homework.script_file else "/python/test.py") return file.decode("utf-8") def on_off_control(af_job_id: int,status: int): for i in range(0,11): parsed_res = get_job_last_parsed_time(af_job_id) last_parsed_time = parsed_res['data']['last_parsed_time'] if last_parsed_time: send_pause(af_job_id,status) print(f"{af_job_id}<==状态修改成功==>{last_parsed_time}") break if i >= 10: raise Exception(f"{af_job_id}==>状态修改失败") time.sleep(2) def execute_job(af_job_id: int): res = get_job_last_parsed_time(af_job_id) current_time = res['data']['last_parsed_time'] if 'last_parsed_time' in res['data'].keys() else None send_submit(af_job_id) for i in range(0,21): parsed_res = get_job_last_parsed_time(af_job_id) last_parsed_time = parsed_res['data']['last_parsed_time'] if last_parsed_time and last_parsed_time != current_time: res = send_execute(af_job_id) print(f"{af_job_id}<==任务执行成功==>{last_parsed_time}") return res if i >= 20: raise Exception(f"{af_job_id}==>文件正在转化中") time.sleep(2)