|
|
@@ -0,0 +1,732 @@
|
|
|
+# -*- coding: utf-8 -*-
|
|
|
+"""PAI 广告模型更新工作流 v5_2（piaoquan_ad_rank_dnn_v15_2）：训练用最近一个月，评估用最近一天。"""
|
|
|
+import functools
|
|
|
+import os
|
|
|
+import re
|
|
|
+import sys
|
|
|
+import time
|
|
|
+import json
|
|
|
+from alibabacloud_paistudio20210202.client import Client as PaiStudio20210202Client
|
|
|
+from alibabacloud_tea_openapi import models as open_api_models
|
|
|
+from alibabacloud_paistudio20210202 import models as pai_studio_20210202_models
|
|
|
+from alibabacloud_tea_util import models as util_models
|
|
|
+from alibabacloud_tea_util.client import Client as UtilClient
|
|
|
+from alibabacloud_eas20210701.client import Client as eas20210701Client
|
|
|
+from alibabacloud_paiflow20210202 import models as paiflow_20210202_models
|
|
|
+from alibabacloud_paiflow20210202.client import Client as PAIFlow20210202Client
|
|
|
+from datetime import datetime, timedelta
|
|
|
+from odps import ODPS
|
|
|
+from ad_monitor_util import _monitor
|
|
|
+import alibabacloud_oss_v2 as oss
|
|
|
+
|
|
|
+target_names = {
|
|
|
+ '样本shuffle',
|
|
|
+ '评估shuffle',
|
|
|
+ '生成CID文件',
|
|
|
+ '模型训练-样本shufle',
|
|
|
+ '模型导出-2',
|
|
|
+ '更新EAS服务(Beta)-1',
|
|
|
+ '虚拟起始节点',
|
|
|
+ '二分类评估-1',
|
|
|
+ '二分类评估-2',
|
|
|
+ '预测结果对比'
|
|
|
+}
|
|
|
+
|
|
|
+WORKFLOW_NAME = "piaoquan_ad_rank_dnn_v15_2"
|
|
|
+EXPERIMENT_ID = "draft-uf7vnc5h5wygurixam"
|
|
|
+ACCESS_KEY_ID = "LTAI5tFGqgC8f3mh1fRCrAEy"
|
|
|
+ACCESS_KEY_SECRET = "XhOjK9XmTYRhVAtf6yii4s4kZwWzvV"
|
|
|
+
|
|
|
+MAX_RETRIES = 3
|
|
|
+TRAIN_DAYS = 30
|
|
|
+EVAL_OFFSET_DAYS = 1
|
|
|
+ONLINE_SERVICE_NAME = 'ad_rank_dnn_v11_easyrec_v6'
|
|
|
+
|
|
|
+
|
|
|
+def retry(func):
|
|
|
+ @functools.wraps(func)
|
|
|
+ def wrapper(*args, **kwargs):
|
|
|
+ retries = 0
|
|
|
+ while retries < MAX_RETRIES:
|
|
|
+ try:
|
|
|
+ result = func(*args, **kwargs)
|
|
|
+ if result is not False:
|
|
|
+ return result
|
|
|
+ except Exception as e:
|
|
|
+ print(f"函数 {func.__name__} 执行时发生异常: {e}，重试第 {retries + 1} 次")
|
|
|
+ retries += 1
|
|
|
+ print(f"函数 {func.__name__} 重试 {MAX_RETRIES} 次后仍失败。")
|
|
|
+ return False
|
|
|
+
|
|
|
+ return wrapper
|
|
|
+
|
|
|
+
|
|
|
+def get_odps_instance(project):
|
|
|
+ odps = ODPS(
|
|
|
+ access_id=ACCESS_KEY_ID,
|
|
|
+ secret_access_key=ACCESS_KEY_SECRET,
|
|
|
+ project=project,
|
|
|
+ endpoint='http://service.cn.maxcompute.aliyun.com/api',
|
|
|
+ )
|
|
|
+ return odps
|
|
|
+
|
|
|
+
|
|
|
+def get_data_from_odps(project, table, num):
|
|
|
+ odps = get_odps_instance(project)
|
|
|
+ try:
|
|
|
+ sql = f'select * from {table} limit {num}'
|
|
|
+ with odps.execute_sql(sql).open_reader() as reader:
|
|
|
+ df = reader.to_pandas()
|
|
|
+ if len(df) < num:
|
|
|
+ return None
|
|
|
+ return df
|
|
|
+ except Exception as e:
|
|
|
+ print(f"发生错误: {e}")
|
|
|
+
|
|
|
+
|
|
|
+def get_dict_from_odps(project, table):
|
|
|
+ odps = get_odps_instance(project)
|
|
|
+ try:
|
|
|
+ sql = f'select * from {table}'
|
|
|
+ with odps.execute_sql(sql).open_reader() as reader:
|
|
|
+ data = {}
|
|
|
+ for record in reader:
|
|
|
+ record_list = list(record)
|
|
|
+ key = record_list[0][1]
|
|
|
+ value = record_list[1][1]
|
|
|
+ data[key] = value
|
|
|
+ return data
|
|
|
+ except Exception as e:
|
|
|
+ print(f"发生错误: {e}")
|
|
|
+
|
|
|
+
|
|
|
+def load_holiday_dates():
|
|
|
+ current_dir = os.getcwd()
|
|
|
+ file_path = os.path.join(current_dir, 'ad', 'holidays.txt')
|
|
|
+ try:
|
|
|
+ with open(file_path, 'r', encoding='utf-8') as file:
|
|
|
+ dates = set()
|
|
|
+ for line in file:
|
|
|
+ token = line.strip()
|
|
|
+ if re.fullmatch(r'\d{8}', token):
|
|
|
+ dates.add(token)
|
|
|
+ return dates
|
|
|
+ except FileNotFoundError:
|
|
|
+ raise Exception(f"错误：未找到 {file_path} 文件。")
|
|
|
+ except Exception as e:
|
|
|
+ raise Exception(f"错误：读取节假日文件失败: {e}")
|
|
|
+
|
|
|
+
|
|
|
+def yyyymmdd_ago(days):
|
|
|
+ return (datetime.now() - timedelta(days=days)).strftime('%Y%m%d')
|
|
|
+
|
|
|
+
|
|
|
+def get_eval_date():
|
|
|
+ """评估日期：昨天 1 天。"""
|
|
|
+ return yyyymmdd_ago(EVAL_OFFSET_DAYS)
|
|
|
+
|
|
|
+
|
|
|
+def get_eval_dates():
|
|
|
+ """评估shuffle：昨天 1 天。"""
|
|
|
+ dates = [get_eval_date()]
|
|
|
+ print(f"v5_2 评估shuffle日期(1天): {dates}")
|
|
|
+ return dates
|
|
|
+
|
|
|
+
|
|
|
+def get_train_dates():
|
|
|
+ """样本shuffle：评估日（昨天）之前再往前取 TRAIN_DAYS 个非节假日，不与评估数据重叠。"""
|
|
|
+ holidays = load_holiday_dates()
|
|
|
+ dates = []
|
|
|
+ offset = EVAL_OFFSET_DAYS + 1
|
|
|
+ while len(dates) < TRAIN_DAYS:
|
|
|
+ day = yyyymmdd_ago(offset)
|
|
|
+ if day not in holidays:
|
|
|
+ dates.append(day)
|
|
|
+ offset += 1
|
|
|
+ if offset > 120:
|
|
|
+ raise Exception(f"无法凑齐 {TRAIN_DAYS} 天训练日期，请检查 holidays.txt")
|
|
|
+ dates.sort()
|
|
|
+ print(f"v5_2 样本shuffle日期({TRAIN_DAYS}天): {dates}")
|
|
|
+ return dates
|
|
|
+
|
|
|
+
|
|
|
+def replace_sql_dt_in(sql, dates):
|
|
|
+ """用指定日期列表替换 SQL 中第一次出现的 where dt in (...)。"""
|
|
|
+ quoted = ','.join(f"'{d}'" for d in dates)
|
|
|
+ marker = 'where dt in ('
|
|
|
+ start_index = sql.find(marker)
|
|
|
+ if start_index == -1:
|
|
|
+ return None
|
|
|
+ value_start = start_index + len(marker)
|
|
|
+ value_end = sql.find(')', value_start)
|
|
|
+ if value_end == -1:
|
|
|
+ return None
|
|
|
+ return sql[:value_start] + quoted + sql[value_end:]
|
|
|
+
|
|
|
+
|
|
|
+def is_created_today(time_str):
|
|
|
+ time_obj = datetime.fromisoformat(time_str)
|
|
|
+ today_start = datetime.combine(datetime.now().date(), datetime.min.time())
|
|
|
+ return time_obj.timestamp() > today_start.timestamp()
|
|
|
+
|
|
|
+
|
|
|
+def replace_odps_table_arg(cmd, flag, table):
|
|
|
+ odps_table = 'odps://pai_algo/tables/' + table
|
|
|
+ marker = f'-D{flag}="'
|
|
|
+ start_index = cmd.find(marker)
|
|
|
+ if start_index == -1:
|
|
|
+ return None
|
|
|
+ value_start = start_index + len(marker)
|
|
|
+ value_end = cmd.find('"', value_start)
|
|
|
+ if value_end == -1:
|
|
|
+ return None
|
|
|
+ return cmd[:value_start] + odps_table + cmd[value_end:]
|
|
|
+
|
|
|
+
|
|
|
+class PAIClient:
|
|
|
+ def __init__(self):
|
|
|
+ pass
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def create_client() -> PaiStudio20210202Client:
|
|
|
+ config = open_api_models.Config(
|
|
|
+ access_key_id=ACCESS_KEY_ID,
|
|
|
+ access_key_secret=ACCESS_KEY_SECRET
|
|
|
+ )
|
|
|
+ config.endpoint = f'pai.cn-hangzhou.aliyuncs.com'
|
|
|
+ return PaiStudio20210202Client(config)
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def create_eas_client() -> eas20210701Client:
|
|
|
+ config = open_api_models.Config(
|
|
|
+ access_key_id=ACCESS_KEY_ID,
|
|
|
+ access_key_secret=ACCESS_KEY_SECRET
|
|
|
+ )
|
|
|
+ config.endpoint = f'pai-eas.cn-hangzhou.aliyuncs.com'
|
|
|
+ return eas20210701Client(config)
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def create_flow_client() -> PAIFlow20210202Client:
|
|
|
+ config = open_api_models.Config(
|
|
|
+ access_key_id=ACCESS_KEY_ID,
|
|
|
+ access_key_secret=ACCESS_KEY_SECRET
|
|
|
+ )
|
|
|
+ config.endpoint = f'paiflow.cn-hangzhou.aliyuncs.com'
|
|
|
+ return PAIFlow20210202Client(config)
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def get_work_flow_draft(experiment_id: str):
|
|
|
+ client = PAIClient.create_client()
|
|
|
+ runtime = util_models.RuntimeOptions()
|
|
|
+ headers = {}
|
|
|
+ try:
|
|
|
+ resp = client.get_experiment_with_options(experiment_id, headers, runtime)
|
|
|
+ return resp.body.to_map()
|
|
|
+ except Exception as error:
|
|
|
+ raise Exception(f"get_work_flow_draft error {error}")
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def get_describe_service(service_name: str):
|
|
|
+ client = PAIClient.create_eas_client()
|
|
|
+ runtime = util_models.RuntimeOptions()
|
|
|
+ headers = {}
|
|
|
+ try:
|
|
|
+ resp = client.describe_service_with_options('cn-hangzhou', service_name, headers, runtime)
|
|
|
+ return resp.body.to_map()
|
|
|
+ except Exception as error:
|
|
|
+ raise Exception(f"get_describe_service error {error}")
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def update_experiment_content(experiment_id: str, content: str, version: int):
|
|
|
+ client = PAIClient.create_client()
|
|
|
+ update_experiment_content_request = pai_studio_20210202_models.UpdateExperimentContentRequest(
|
|
|
+ content=content, version=version)
|
|
|
+ runtime = util_models.RuntimeOptions()
|
|
|
+ headers = {}
|
|
|
+ try:
|
|
|
+ resp = client.update_experiment_content_with_options(
|
|
|
+ experiment_id, update_experiment_content_request, headers, runtime)
|
|
|
+ print(resp.body.to_map())
|
|
|
+ except Exception as error:
|
|
|
+ raise Exception(f"update_experiment_content error {error}")
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def create_job(experiment_id: str, node_id: str, execute_type: str):
|
|
|
+ client = PAIClient.create_client()
|
|
|
+ create_job_request = pai_studio_20210202_models.CreateJobRequest()
|
|
|
+ create_job_request.experiment_id = experiment_id
|
|
|
+ create_job_request.node_id = node_id
|
|
|
+ create_job_request.execute_type = execute_type
|
|
|
+ runtime = util_models.RuntimeOptions()
|
|
|
+ headers = {}
|
|
|
+ try:
|
|
|
+ resp = client.create_job_with_options(create_job_request, headers, runtime)
|
|
|
+ return resp.body.to_map()
|
|
|
+ except Exception as error:
|
|
|
+ raise Exception(f"create_job error {error}")
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def get_jobs_list(experiment_id: str, order='DESC'):
|
|
|
+ client = PAIClient.create_client()
|
|
|
+ list_jobs_request = pai_studio_20210202_models.ListJobsRequest(
|
|
|
+ experiment_id=experiment_id,
|
|
|
+ order=order
|
|
|
+ )
|
|
|
+ runtime = util_models.RuntimeOptions()
|
|
|
+ headers = {}
|
|
|
+ try:
|
|
|
+ resp = client.list_jobs_with_options(list_jobs_request, headers, runtime)
|
|
|
+ return resp.body.to_map()
|
|
|
+ except Exception as error:
|
|
|
+ raise Exception(f"get_jobs_list error {error}")
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def get_job_detail(job_id: str, verbose=False):
|
|
|
+ client = PAIClient.create_client()
|
|
|
+ get_job_request = pai_studio_20210202_models.GetJobRequest(
|
|
|
+ verbose=verbose
|
|
|
+ )
|
|
|
+ runtime = util_models.RuntimeOptions()
|
|
|
+ headers = {}
|
|
|
+ try:
|
|
|
+ resp = client.get_job_with_options(job_id, get_job_request, headers, runtime)
|
|
|
+ return resp.body.to_map()
|
|
|
+ except Exception as error:
|
|
|
+ print(error.message)
|
|
|
+ print(error.data.get("Recommend"))
|
|
|
+ UtilClient.assert_as_string(error.message)
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def get_flow_out_put(pipeline_run_id: str, node_id: str, depth: int):
|
|
|
+ client = PAIClient.create_flow_client()
|
|
|
+ list_pipeline_run_node_outputs_request = paiflow_20210202_models.ListPipelineRunNodeOutputsRequest(
|
|
|
+ depth=depth
|
|
|
+ )
|
|
|
+ runtime = util_models.RuntimeOptions()
|
|
|
+ headers = {}
|
|
|
+ try:
|
|
|
+ resp = client.list_pipeline_run_node_outputs_with_options(
|
|
|
+ pipeline_run_id, node_id, list_pipeline_run_node_outputs_request, headers, runtime)
|
|
|
+ return resp.body.to_map()
|
|
|
+ except Exception as error:
|
|
|
+ print(error.message)
|
|
|
+ print(error.data.get("Recommend"))
|
|
|
+ UtilClient.assert_as_string(error.message)
|
|
|
+
|
|
|
+
|
|
|
+def extract_date_yyyymmdd(input_string):
|
|
|
+ pattern = r'\d{8}'
|
|
|
+ matches = re.findall(pattern, input_string)
|
|
|
+ if matches:
|
|
|
+ return matches[0]
|
|
|
+ return None
|
|
|
+
|
|
|
+
|
|
|
+def get_online_model_config(service_name: str):
|
|
|
+ model_config = {}
|
|
|
+ model_detail = PAIClient.get_describe_service(service_name)
|
|
|
+ service_config_str = model_detail['ServiceConfig']
|
|
|
+ service_config = json.loads(service_config_str)
|
|
|
+ model_path = service_config['model_path']
|
|
|
+ model_config['model_path'] = model_path
|
|
|
+ model_config['online_date'] = extract_date_yyyymmdd(model_path)
|
|
|
+ return model_config
|
|
|
+
|
|
|
+
|
|
|
+def get_shuffle_output_table(node_name, node_dict, job_dict):
|
|
|
+ job_id = job_dict[node_name]
|
|
|
+ job_detail = wait_job_end(job_id)
|
|
|
+ if job_detail['Status'] != 'Succeeded':
|
|
|
+ return None
|
|
|
+ flow_out_put_detail = PAIClient.get_flow_out_put(job_detail['RunId'], job_detail['PaiflowNodeId'], 2)
|
|
|
+ outputs = flow_out_put_detail['Outputs']
|
|
|
+ for output in outputs:
|
|
|
+ if output["Producer"] == node_dict[node_name] and output["Name"] == "outputTable":
|
|
|
+ value = json.loads(output["Info"]['value'])
|
|
|
+ table = value['location']['table']
|
|
|
+ print(f"{node_name} outputTable: {table}")
|
|
|
+ return table
|
|
|
+ return None
|
|
|
+
|
|
|
+
|
|
|
+def bind_shuffle_tables(train_table, eval_table):
|
|
|
+ draft = PAIClient.get_work_flow_draft(EXPERIMENT_ID)
|
|
|
+ print(json.dumps(draft, ensure_ascii=False))
|
|
|
+ content = draft['Content']
|
|
|
+ version = draft['Version']
|
|
|
+ content_json = json.loads(content)
|
|
|
+ nodes = content_json.get('nodes')
|
|
|
+ predict_node_names = {'模型预测', '线上模型预测'}
|
|
|
+ for node in nodes:
|
|
|
+ name = node['name']
|
|
|
+ for property in node['properties']:
|
|
|
+ if property['name'] != 'sql':
|
|
|
+ continue
|
|
|
+ cmd = property['value']
|
|
|
+ if name == '模型训练-样本shufle':
|
|
|
+ new_cmd = replace_odps_table_arg(cmd, 'train_tables', train_table)
|
|
|
+ if new_cmd is None:
|
|
|
+ print("replace Dtrain_tables error")
|
|
|
+ new_cmd = cmd
|
|
|
+ new_cmd = replace_odps_table_arg(new_cmd, 'eval_tables', eval_table)
|
|
|
+ if new_cmd is None:
|
|
|
+ print("replace Deval_tables error")
|
|
|
+ property['value'] = new_cmd
|
|
|
+ elif name in predict_node_names:
|
|
|
+ new_cmd = replace_odps_table_arg(cmd, 'input_table', train_table)
|
|
|
+ if new_cmd is None:
|
|
|
+ print(f"replace Dinput_table error for {name}")
|
|
|
+ else:
|
|
|
+ property['value'] = new_cmd
|
|
|
+ print(f"{name} Dinput_table -> {train_table}")
|
|
|
+ new_content = json.dumps(content_json, ensure_ascii=False)
|
|
|
+ PAIClient.update_experiment_content(EXPERIMENT_ID, new_content, version)
|
|
|
+
|
|
|
+
|
|
|
+def wait_job_end(job_id: str, check_interval=300):
|
|
|
+ while True:
|
|
|
+ job_detail = PAIClient.get_job_detail(job_id)
|
|
|
+ print(job_detail)
|
|
|
+ statue = job_detail['Status']
|
|
|
+ if (statue == 'Initialized' or statue == 'Starting' or statue == 'WorkflowServiceStarting'
|
|
|
+ or statue == 'Running' or statue == 'ReadyToSchedule'):
|
|
|
+ time.sleep(check_interval)
|
|
|
+ continue
|
|
|
+ if statue == 'Failed' or statue == 'Terminating' or statue == 'Unknown' or statue == 'Skipped' or statue == 'Succeeded':
|
|
|
+ return job_detail
|
|
|
+
|
|
|
+
|
|
|
+def get_node_dict():
|
|
|
+ draft = PAIClient.get_work_flow_draft(EXPERIMENT_ID)
|
|
|
+ content = draft['Content']
|
|
|
+ content_json = json.loads(content)
|
|
|
+ nodes = content_json.get('nodes')
|
|
|
+ node_dict = {}
|
|
|
+ for node in nodes:
|
|
|
+ name = node['name']
|
|
|
+ if name in target_names:
|
|
|
+ node_dict[name] = node['id']
|
|
|
+ return node_dict
|
|
|
+
|
|
|
+
|
|
|
+def get_job_dict():
|
|
|
+ job_dict = {}
|
|
|
+ jobs_list = PAIClient.get_jobs_list(EXPERIMENT_ID)
|
|
|
+ for job in jobs_list['Jobs']:
|
|
|
+ if not is_created_today(job['GmtCreateTime']):
|
|
|
+ break
|
|
|
+ job_id = job['JobId']
|
|
|
+ job_detail = PAIClient.get_job_detail(job_id, verbose=True)
|
|
|
+ for name in target_names:
|
|
|
+ if job_detail['Status'] != 'Succeeded':
|
|
|
+ continue
|
|
|
+ if name in job_dict:
|
|
|
+ continue
|
|
|
+ if name in job_detail['RunInfo']:
|
|
|
+ job_dict[name] = job_detail['JobId']
|
|
|
+ return job_dict
|
|
|
+
|
|
|
+
|
|
|
+@retry
|
|
|
+def update_online_flow():
|
|
|
+ try:
|
|
|
+ online_model_config = get_online_model_config(ONLINE_SERVICE_NAME)
|
|
|
+ draft = PAIClient.get_work_flow_draft(EXPERIMENT_ID)
|
|
|
+ print(json.dumps(draft, ensure_ascii=False))
|
|
|
+ content = draft['Content']
|
|
|
+ version = draft['Version']
|
|
|
+ print(content)
|
|
|
+ content_json = json.loads(content)
|
|
|
+ nodes = content_json.get('nodes')
|
|
|
+ global_params = content_json.get('globalParams')
|
|
|
+ eval_date = get_eval_date()
|
|
|
+ train_dates = get_train_dates()
|
|
|
+ eval_dates = get_eval_dates()
|
|
|
+ for global_param in global_params:
|
|
|
+ try:
|
|
|
+ if global_param['name'] == 'bizdate':
|
|
|
+ global_param['value'] = eval_date
|
|
|
+ if global_param['name'] == 'online_version_dt':
|
|
|
+ global_param['value'] = online_model_config['online_date']
|
|
|
+ if global_param['name'] == 'eval_date':
|
|
|
+ global_param['value'] = eval_date
|
|
|
+ if global_param['name'] == 'online_model_path':
|
|
|
+ global_param['value'] = online_model_config['model_path']
|
|
|
+ except KeyError:
|
|
|
+ raise Exception("在处理全局参数时，字典中缺少必要的键")
|
|
|
+ shuffle_dates = {
|
|
|
+ '样本shuffle': train_dates,
|
|
|
+ '评估shuffle': eval_dates,
|
|
|
+ }
|
|
|
+ for node in nodes:
|
|
|
+ try:
|
|
|
+ if node['name'] not in shuffle_dates:
|
|
|
+ continue
|
|
|
+ for property in node['properties']:
|
|
|
+ if property['name'] != 'sql':
|
|
|
+ continue
|
|
|
+ new_value = replace_sql_dt_in(property['value'], shuffle_dates[node['name']])
|
|
|
+ if new_value is None:
|
|
|
+ print(f"error replace dt for {node['name']}")
|
|
|
+ property['value'] = new_value
|
|
|
+ except KeyError:
|
|
|
+ raise Exception("在处理节点属性时，字典中缺少必要的键")
|
|
|
+ new_content = json.dumps(content_json, ensure_ascii=False)
|
|
|
+ PAIClient.update_experiment_content(EXPERIMENT_ID, new_content, version)
|
|
|
+ return True
|
|
|
+ except json.JSONDecodeError:
|
|
|
+ raise Exception("JSON 解析错误，可能是草稿内容格式不正确")
|
|
|
+ except Exception as e:
|
|
|
+ raise Exception(f"发生未知错误: {e}")
|
|
|
+
|
|
|
+
|
|
|
+@retry
|
|
|
+def shuffle_table():
|
|
|
+ try:
|
|
|
+ node_dict = get_node_dict()
|
|
|
+ if '生成CID文件' not in node_dict:
|
|
|
+ raise Exception("工作流中未找到节点 生成CID文件")
|
|
|
+ train_res = PAIClient.create_job(EXPERIMENT_ID, node_dict['样本shuffle'], 'EXECUTE_FROM_HERE')
|
|
|
+ eval_res = PAIClient.create_job(EXPERIMENT_ID, node_dict['评估shuffle'], 'EXECUTE_ONE')
|
|
|
+ train_job_detail = wait_job_end(train_res['JobId'], 10)
|
|
|
+ eval_job_detail = wait_job_end(eval_res['JobId'], 10)
|
|
|
+ if train_job_detail['Status'] != 'Succeeded' or eval_job_detail['Status'] != 'Succeeded':
|
|
|
+ return False
|
|
|
+ job_verbose = PAIClient.get_job_detail(train_res['JobId'], verbose=True)
|
|
|
+ run_info = job_verbose.get('RunInfo') or ''
|
|
|
+ if '生成CID文件' not in run_info:
|
|
|
+ print(f"样本shuffle 未带上生成CID文件, RunInfo={run_info}")
|
|
|
+ return False
|
|
|
+ print("样本shuffle 已触发生成CID文件")
|
|
|
+ return True
|
|
|
+ except Exception as e:
|
|
|
+ error_message = f"在执行 shuffle_table 函数时发生异常: {str(e)}"
|
|
|
+ print(error_message)
|
|
|
+ raise Exception(error_message)
|
|
|
+
|
|
|
+
|
|
|
+@retry
|
|
|
+def shuffle_train_model():
|
|
|
+ try:
|
|
|
+ node_dict = get_node_dict()
|
|
|
+ job_dict = get_job_dict()
|
|
|
+ train_table = get_shuffle_output_table('样本shuffle', node_dict, job_dict)
|
|
|
+ eval_table = get_shuffle_output_table('评估shuffle', node_dict, job_dict)
|
|
|
+ if train_table is None or eval_table is None:
|
|
|
+ print(f"shuffle 输出表缺失 train_table={train_table}, eval_table={eval_table}")
|
|
|
+ return False
|
|
|
+ bind_shuffle_tables(train_table, eval_table)
|
|
|
+ node_dict = get_node_dict()
|
|
|
+ train_node_id = node_dict['模型训练-样本shufle']
|
|
|
+ execute_type = 'EXECUTE_ONE'
|
|
|
+ train_res = PAIClient.create_job(EXPERIMENT_ID, train_node_id, execute_type)
|
|
|
+ train_job_id = train_res['JobId']
|
|
|
+ train_job_detail = wait_job_end(train_job_id)
|
|
|
+ if train_job_detail['Status'] == 'Succeeded':
|
|
|
+ return True
|
|
|
+ return False
|
|
|
+ except Exception as e:
|
|
|
+ error_message = f"在执行 shuffle_train_model 函数时发生异常: {str(e)}"
|
|
|
+ print(error_message)
|
|
|
+ raise Exception(error_message)
|
|
|
+
|
|
|
+
|
|
|
+@retry
|
|
|
+def export_model():
|
|
|
+ try:
|
|
|
+ node_dict = get_node_dict()
|
|
|
+ export_node_id = node_dict['模型导出-2']
|
|
|
+ execute_type = 'EXECUTE_ONE'
|
|
|
+ export_res = PAIClient.create_job(EXPERIMENT_ID, export_node_id, execute_type)
|
|
|
+ export_job_id = export_res['JobId']
|
|
|
+ export_job_detail = wait_job_end(export_job_id)
|
|
|
+ if export_job_detail['Status'] == 'Succeeded':
|
|
|
+ return True
|
|
|
+ return False
|
|
|
+ except Exception as e:
|
|
|
+ error_message = f"在执行 export_model 函数时发生异常: {str(e)}"
|
|
|
+ print(error_message)
|
|
|
+ raise Exception(error_message)
|
|
|
+
|
|
|
+
|
|
|
+def update_online_model():
|
|
|
+ try:
|
|
|
+ node_dict = get_node_dict()
|
|
|
+ train_node_id = node_dict['更新EAS服务(Beta)-1']
|
|
|
+ execute_type = 'EXECUTE_ONE'
|
|
|
+ train_res = PAIClient.create_job(EXPERIMENT_ID, train_node_id, execute_type)
|
|
|
+ train_job_id = train_res['JobId']
|
|
|
+ train_job_detail = wait_job_end(train_job_id)
|
|
|
+ if train_job_detail['Status'] == 'Succeeded':
|
|
|
+ return True
|
|
|
+ return False
|
|
|
+ except Exception as e:
|
|
|
+ error_message = f"在执行 update_online_model 函数时发生异常: {str(e)}"
|
|
|
+ print(error_message)
|
|
|
+ raise Exception(error_message)
|
|
|
+
|
|
|
+
|
|
|
+@retry
|
|
|
+def get_validate_model_data():
|
|
|
+ try:
|
|
|
+ node_dict = get_node_dict()
|
|
|
+ train_node_id = node_dict['虚拟起始节点']
|
|
|
+ execute_type = 'EXECUTE_FROM_HERE'
|
|
|
+ validate_res = PAIClient.create_job(EXPERIMENT_ID, train_node_id, execute_type)
|
|
|
+ validate_job_id = validate_res['JobId']
|
|
|
+ validate_job_detail = wait_job_end(validate_job_id)
|
|
|
+ if validate_job_detail['Status'] == 'Succeeded':
|
|
|
+ return True
|
|
|
+ return False
|
|
|
+ except Exception as e:
|
|
|
+ error_message = f"在执行 get_validate_model_data 函数时出现异常: {e}"
|
|
|
+ print(error_message)
|
|
|
+ raise Exception(error_message)
|
|
|
+
|
|
|
+
|
|
|
+def validate_model_data_accuracy():
|
|
|
+ try:
|
|
|
+ table_dict = {}
|
|
|
+ node_dict = get_node_dict()
|
|
|
+ job_dict = get_job_dict()
|
|
|
+ job_id = job_dict['虚拟起始节点']
|
|
|
+ validate_job_detail = wait_job_end(job_id)
|
|
|
+ if validate_job_detail['Status'] == 'Succeeded':
|
|
|
+ pipeline_run_id = validate_job_detail['RunId']
|
|
|
+ node_id = validate_job_detail['PaiflowNodeId']
|
|
|
+ flow_out_put_detail = PAIClient.get_flow_out_put(pipeline_run_id, node_id, 3)
|
|
|
+ print(flow_out_put_detail)
|
|
|
+ outputs = flow_out_put_detail['Outputs']
|
|
|
+ for output in outputs:
|
|
|
+ if output["Producer"] == node_dict['二分类评估-1'] and output["Name"] == "outputMetricTable":
|
|
|
+ value1 = json.loads(output["Info"]['value'])
|
|
|
+ table_dict['二分类评估-1'] = value1['location']['table']
|
|
|
+ if output["Producer"] == node_dict['二分类评估-2'] and output["Name"] == "outputMetricTable":
|
|
|
+ value2 = json.loads(output["Info"]['value'])
|
|
|
+ table_dict['二分类评估-2'] = value2['location']['table']
|
|
|
+ if output["Producer"] == node_dict['预测结果对比'] and output["Name"] == "outputTable":
|
|
|
+ value3 = json.loads(output["Info"]['value'])
|
|
|
+ table_dict['预测结果对比'] = value3['location']['table']
|
|
|
+ num = 10
|
|
|
+ df = get_data_from_odps('pai_algo', table_dict['预测结果对比'], 10)
|
|
|
+ old_abs_avg = df['old_error'].abs().sum() / num
|
|
|
+ new_abs_avg = df['new_error'].abs().sum() / num
|
|
|
+ new_auc = get_dict_from_odps('pai_algo', table_dict['二分类评估-1'])['AUC']
|
|
|
+ old_auc = get_dict_from_odps('pai_algo', table_dict['二分类评估-2'])['AUC']
|
|
|
+ eval_date = get_eval_date()
|
|
|
+ score_diff = abs(old_abs_avg - new_abs_avg)
|
|
|
+ msg = ""
|
|
|
+ result = False
|
|
|
+ if new_abs_avg > 0.1:
|
|
|
+ msg += f'{WORKFLOW_NAME}线上模型评估{eval_date}的数据，绝对误差大于0.1，请检查'
|
|
|
+ level = 'error'
|
|
|
+ elif score_diff > 0.02 and new_abs_avg - old_abs_avg > 0.02:
|
|
|
+ msg += f'{WORKFLOW_NAME}两个模型评估{eval_date}的数据，两个模型分数差异为: {score_diff}, 大于0.02, 请检查'
|
|
|
+ level = 'error'
|
|
|
+ else:
|
|
|
+ msg += f'{WORKFLOW_NAME}广告模型更新完成(训练30天/评估1天)'
|
|
|
+ level = 'info'
|
|
|
+ result = True
|
|
|
+
|
|
|
+ top10_msg = "| CID | 老模型相对真实CTCVR的变化 | 新模型相对真实CTCVR的变化 |"
|
|
|
+ top10_msg += "\n| ---- | --------- | -------- |"
|
|
|
+
|
|
|
+ for index, row in df.iterrows():
|
|
|
+ cid = row['cid']
|
|
|
+ old_error = row['old_error']
|
|
|
+ new_error = row['new_error']
|
|
|
+ top10_msg += f"\n| {int(cid)} | {old_error} | {new_error} | "
|
|
|
+ print(top10_msg)
|
|
|
+ msg += f"\n\t - 老模型AUC: {old_auc}"
|
|
|
+ msg += f"\n\t - 新模型AUC: {new_auc}"
|
|
|
+ msg += f"\n\t - 老模型Top10差异平均值: {old_abs_avg}"
|
|
|
+ msg += f"\n\t - 新模型Top10差异平均值: {new_abs_avg}"
|
|
|
+ return result, msg, level, top10_msg
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ error_message = f"在执行 validate_model_data_accuracy 函数时出现异常: {str(e)}"
|
|
|
+ print(error_message)
|
|
|
+ raise Exception(error_message)
|
|
|
+
|
|
|
+
|
|
|
+def update_trained_cids_pointer(model_name=None, dt_version=None):
|
|
|
+ if not model_name and not dt_version:
|
|
|
+ draft = PAIClient.get_work_flow_draft(EXPERIMENT_ID)
|
|
|
+ content = draft['Content']
|
|
|
+ content_json = json.loads(content)
|
|
|
+ global_params = content_json.get('globalParams', [])
|
|
|
+ model_name = None
|
|
|
+ dt_version = None
|
|
|
+ for param in global_params:
|
|
|
+ if param.get('name') == 'model_name':
|
|
|
+ model_name = param.get('value')
|
|
|
+ if param.get('name') == 'bizdate':
|
|
|
+ dt_version = param.get('value')
|
|
|
+ if not model_name or not dt_version:
|
|
|
+ raise Exception("globalParams 中未找到 model_name 或 bizdate")
|
|
|
+ elif not (model_name and dt_version):
|
|
|
+ raise Exception("model_name 和 dt_version 必须同时提供")
|
|
|
+ model_version = {}
|
|
|
+ model_version['modelName'] = f"model_name={model_name}"
|
|
|
+ model_version['dtVersion'] = f"dt_version={dt_version}"
|
|
|
+ model_version['timestamp'] = int(time.time())
|
|
|
+ print(json.dumps(model_version, ensure_ascii=False, indent=4).encode('utf-8'))
|
|
|
+ bucket_name = "art-recommend"
|
|
|
+ object_key = "fengzhoutian/pai_model_trained_cids/model_version_v5_2.json"
|
|
|
+
|
|
|
+ oss_config = oss.config.load_default()
|
|
|
+ oss_config.credentials_provider = oss.credentials.StaticCredentialsProvider(
|
|
|
+ access_key_id=ACCESS_KEY_ID, access_key_secret=ACCESS_KEY_SECRET
|
|
|
+ )
|
|
|
+ oss_config.region = "cn-hangzhou"
|
|
|
+ client = oss.Client(oss_config)
|
|
|
+ ret = client.put_object(oss.PutObjectRequest(
|
|
|
+ bucket=bucket_name,
|
|
|
+ key=object_key,
|
|
|
+ body=json.dumps(model_version, ensure_ascii=False, indent=4).encode('utf-8')
|
|
|
+ ))
|
|
|
+ print(f'oss put status code: {ret.status_code},'
|
|
|
+ f' request id: {ret.request_id},'
|
|
|
+ f' content md5: {ret.content_md5},'
|
|
|
+ f' etag: {ret.etag},'
|
|
|
+ f' hash crc64: {ret.hash_crc64},'
|
|
|
+ f' version id: {ret.version_id},'
|
|
|
+ f' content: {model_version}'
|
|
|
+ )
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == '__main__':
|
|
|
+ start_time = int(time.time())
|
|
|
+ functions = [update_online_flow, shuffle_table, shuffle_train_model, export_model, get_validate_model_data]
|
|
|
+ function_names = [func.__name__ for func in functions]
|
|
|
+
|
|
|
+ start_function = None
|
|
|
+ if len(sys.argv) > 1:
|
|
|
+ start_function = sys.argv[1]
|
|
|
+ if start_function not in function_names:
|
|
|
+ print(f"指定的起始函数 {start_function} 不存在，请选择以下函数之一：{', '.join(function_names)}")
|
|
|
+ sys.exit(1)
|
|
|
+
|
|
|
+ start_index = 0
|
|
|
+ if start_function:
|
|
|
+ start_index = function_names.index(start_function)
|
|
|
+
|
|
|
+ for func in functions[start_index:]:
|
|
|
+ if not func():
|
|
|
+ print(f"{func.__name__} 执行失败，后续函数不再执行。")
|
|
|
+ step_end_time = int(time.time())
|
|
|
+ elapsed = step_end_time - start_time
|
|
|
+ _monitor('error', f"{WORKFLOW_NAME}模型更新，{func.__name__} 执行失败，后续函数不再执行，请检查", start_time, elapsed, None)
|
|
|
+ break
|
|
|
+ else:
|
|
|
+ print("所有函数都成功执行，可以继续下一步操作。")
|
|
|
+ result, msg, level, top10_msg = validate_model_data_accuracy()
|
|
|
+ if result:
|
|
|
+ update_online_res = update_online_model()
|
|
|
+ if update_online_res:
|
|
|
+ update_trained_cids_pointer()
|
|
|
+ print("success")
|
|
|
+ step_end_time = int(time.time())
|
|
|
+ elapsed = step_end_time - start_time
|
|
|
+ print(level, msg, start_time, elapsed, top10_msg)
|
|
|
+ _monitor(level, msg, start_time, elapsed, top10_msg)
|
xueyiming