diff --git a/clean_job.yaml b/clean_job.yaml new file mode 100644 index 0000000..00422a1 --- /dev/null +++ b/clean_job.yaml @@ -0,0 +1,45 @@ +apiVersion: batch/v1 +kind: Job +metadata: + name: delete-old-rows + namespace: sientia +spec: + template: + spec: + containers: + - name: delete-old-rows + image: docker.io/bitnami/postgresql:16.2.0-debian-12-r10 + env: + - name: PGPASSWORD + value: "asidhsd@!#!@@!ASD!@#!ASDQ@#!FSDTRYJG#@@$#@%" + - name: PGUSER + value: temporal + - name: PGHOST + value: "paradedb-rw.paradedb.svc.cluster.local" + - name: PGDATABASE + value: "temporal_visibility" + command: + - "sh" + - "-c" + - | + # COMANDO CORRIGIDO - Excluir apenas workflows COMPLETED/FAILED antigos + # Preserva schedules (que ficam RUNNING) e workflows recentes + psql -h $PGHOST -U $PGUSER -d $PGDATABASE -c " + DELETE FROM public.executions_visibility + WHERE start_time < NOW() - INTERVAL '1 minutes' + AND status IN (2, 3, 4, 5, 7);" + + # Comando para executar VACUUM FULL após a exclusão + psql -h $PGHOST -U $PGUSER -d $PGDATABASE -c "VACUUM FULL public.executions_visibility;" + envFrom: + - secretRef: + name: postgres-credentials + restartPolicy: Never + backoffLimit: 0 + ttlSecondsAfterFinished: 3600 + +# kubectl apply -f clean_job.yaml -n sientia + +# kubectl create secret generic postgres-credentials --from-literal=postgres-password=sientia --from-literal=postgres-username=sientia -n sientia4 + +# drop database temporal; drop database temporal_visibility; create database temporal owner temporal; create database temporal_visibility owner temporal; \ No newline at end of file diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index 4a71455..832cfbb 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -207,13 +207,12 @@ class Gates(BaseActivity): filter_output = [] self.debug( - f"Input data: \n {create_sample_dict(data, max_items=5, max_depth=2)}", metadata) + f"Input data: \n {create_sample_dict(data, max_items=5, max_depth=5)}", metadata) self.debug(f"Filters: {filters}", metadata) comments = [] for fil, config in filters.items(): if fil not in mlflow_response_filter_functions: - self.error(f"Filter {fil} not found", metadata) continue try: if mlflow_response_filter_functions[fil](data, config): @@ -293,7 +292,7 @@ class Gates(BaseActivity): filter_output = [] self.debug(f"Input data:\n {data.head(5).to_string()}", metadata) - self.debug(f"Filters: \n {create_sample_dict(filters)}", metadata) + self.debug(f"Filters: \n {filters}", metadata) for fil, config in filters.items(): if fil not in mlflow_content_filter_functions: @@ -492,6 +491,36 @@ class Gates(BaseActivity): self.info(f"Default prediction formatted: {data.size} rows", metadata) return data.to_dict() + @activity.defn(name="format_retrain_report") + async def format_retrain_report(self, input_data: dict[str, Any]) -> dict[Any, Any]: + """ + Format retrain report data according to configured storage policies. + """ + metadata = input_data['metadata'] + self.info("Formatting retrain report...", metadata) + + experiment_response = input_data['experiment_response'] + update_report = input_data['update_report'] + model_id = input_data['model_id'] + model_name = input_data['model_name'] + + report = DataFrame({ + 'model_id': [model_id], + 'model_name': [model_name], + 'timestamp': [experiment_response['timestamp']], + 'status': [experiment_response['message']] + }) + + if experiment_response['success']: + # Retrain was successfull + report['version'] = update_report['version'] + report['mlflow_run_id'] = update_report['mlflow_run_id'] + report['mlflow_experiment_id'] = update_report['mlflow_experiment_id'] + + self.debug(f"Retrain report: {report.to_csv()}", metadata) + + return report.to_dict() + @activity.defn(name="get_last_timestamp") async def get_last_timestamp(self, input_data: dict[str, Any]) -> str: """ diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index d6dc98d..7ad795b 100644 --- a/laborious/activities/mlflow.py +++ b/laborious/activities/mlflow.py @@ -243,30 +243,29 @@ class MLFlow(BaseActivity): data = data.dropna() data.columns.name = None - try: - retrain_output, experiment = self.model_monitoring_repository.retrain_model( - data=data, - model_name=model_name, - model_config=model_config - ) + retrain_output = self.model_monitoring_repository.retrain_model( + data=data, + model_name=model_name, + model_config=model_config + ) - return { - 'status': retrain_output, - 'timestamp': timestamp, - 'experiment': experiment - } - except Exception as e: - trace = traceback.format_exc() + if not retrain_output['success']: + + trace = retrain_output['traceback'] self.send_notification( metadata=metadata, notification_id='RETRAIN_MODEL_ERROR', - message=f'Error retraining model {model_name}: {e}', + message=f'Error retraining model {model_name}: {retrain_output['message']}', block='retrain_model', level=NotificationLevel.ERROR, attachment_content=trace ) self.error(trace, metadata=metadata) - raise e + + return { + **retrain_output, + 'timestamp': timestamp + } @activity.defn(name="update_production_model") async def update_production_model(self, input_data: dict[str, Any]) -> dict[Any, Any]: @@ -307,11 +306,7 @@ class MLFlow(BaseActivity): """ metadata = input_data['metadata'] model_name = input_data['model_name'] - model_id = input_data['model_id'] experiment = input_data['experiment'] - timestamp = input_data['timestamp'] - status = input_data['status'] - self.info( f'Updating production model {model_name} from experiment {experiment}...', metadata) @@ -321,15 +316,9 @@ class MLFlow(BaseActivity): model_name=model_name ) - report = DataFrame([response]) - report['model_id'] = model_id - report['model_name'] = model_name - report['timestamp'] = timestamp - report['status'] = status - self.info( f'Production model {model_name} updated successfully', metadata) - return report.to_dict() + return response except Exception as e: trace = traceback.format_exc() diff --git a/laborious/utils/filters/conditional_filters.py b/laborious/utils/filters/conditional_filters.py index 2c51805..6caf1e5 100644 --- a/laborious/utils/filters/conditional_filters.py +++ b/laborious/utils/filters/conditional_filters.py @@ -20,6 +20,10 @@ def filter_specific_variables_null_values(data: DataFrame, config: dict) -> bool False if none of the specified variables contain null values. """ + + if data.empty: + return False + return not data[ data['variable'].isin(config['variables']) & data['value'].isna()].empty diff --git a/laborious/utils/repository/model_repository.py b/laborious/utils/repository/model_repository.py index 7ce7b32..1816c80 100644 --- a/laborious/utils/repository/model_repository.py +++ b/laborious/utils/repository/model_repository.py @@ -76,16 +76,30 @@ class MLFlowRepository(): Returns: str: The run_id of the model. """ - latest_versions = self.client.get_latest_versions( - name=model_name, stages=[stage] + # Use search_registered_models instead of deprecated get_latest_versions + registered_models = self.client.search_registered_models( + filter_string=f"name='{model_name}'" ) - if not latest_versions: + + if not registered_models: + raise mlflow.exceptions.MlflowException( + f"Model '{model_name}' not found in the Model Registry." + ) + + # Get the latest version in the specified stage + model_versions = self.client.search_model_versions( + filter_string=f"name='{model_name}' and stage='{stage}'" + ) + + if not model_versions: raise mlflow.exceptions.MlflowException( f"Model '{model_name}' in stage '{stage}' not found in the Model Registry." ) - else: - run_id = latest_versions[0].source.split("/") - return run_id[2] + + # Sort by version number to get the latest + latest_version = max(model_versions, key=lambda v: int(v.version)) + run_id = latest_version.source.split("/") + return run_id[2] def get_experiment_by_run_id(self, run_id: str) -> str: """ @@ -482,7 +496,7 @@ class MLFlowRepository(): Returns: bool: True if cache is still valid, False if expired """ - current_time = self.now() + current_time = datetime.now() cache_time = cache['timestamp'] if current_time - cache_time >= timedelta(minutes=retention): return False @@ -601,7 +615,7 @@ class MLFlowRepository(): cache = { 'target': model_config_to_cache, 'config': config, - 'timestamp': self.now() + 'timestamp': datetime.now() } self.model_cache[model_key] = cache @@ -672,7 +686,7 @@ class MLFlowRepository(): def create_model_experiment(self, model_name: str, data: pd.DataFrame, transform_flavor: str = 'sklearn', predict_flavor: str = 'pyfunc', - compressed: bool = False) -> tuple: + compressed: bool = False, fit_config: dict = {}, target_name: str = None) -> tuple: """ Create a new MLFlow experiment for model retraining. @@ -701,18 +715,34 @@ class MLFlowRepository(): ) data_model = self.download_model( model_name, "transform", transform_flavor, compressed - ) + )['model'] prediction_model = self.download_model( model_name, "predict", predict_flavor, compressed - ) + )['model'] data_model = data_model.fit(data) treated_data = data_model.predict(data) - target_name = data_model.target_variable - y = data[target_name] - treated_data = pd.merge( - treated_data, y, left_index=True, right_index=True) - prediction_model = prediction_model.fit(treated_data) + if target_name is None: + target_name = data_model.target_variable + + if fit_config.get('y_type', 'series').lower() == 'series': + y = treated_data[target_name] + else: + y = treated_data[[target_name]] + + if not fit_config.get('split_fit_data', False): + # Merge treated data with target + treated_data = pd.merge( + treated_data, y, left_index=True, right_index=True) + + prediction_model = prediction_model.fit(treated_data) + else: + # Keep data separated + if fit_config.get('split_fit_first', 'x').lower() == 'x': + prediction_model = prediction_model.fit(treated_data, y) + else: + prediction_model = prediction_model.fit(y, treated_data) + experiment = self.get_experiment_by_run_id(latest_production_id) mlflow.set_experiment(experiment) @@ -781,7 +811,7 @@ class MLFlowRepository(): if path.exists(file_path): remove(file_path) - return "Model retrained successfully", experiment + return experiment def update_production_model_by_run_id(self, run_id: str, model_name: str) -> dict: """ @@ -844,7 +874,7 @@ class MLFlowRepository(): Functions that provide the interface to model operations """ - def transform(self, model_name: str, data: pd.DataFrame, model_retention: int, + def transform(self, model_name: str, data: pd.DataFrame, model_config: dict, metadata: dict): """ Transform data using a cached transformation model. @@ -863,7 +893,6 @@ class MLFlowRepository(): Parameters: model_name (str): The name of the MLFlow model to use for transformation. data (pd.DataFrame): The input data to be transformed by the model. - model_retention (int): Cache retention time in minutes (0 = no caching). model_config (dict): Model configuration parameters metadata (dict): Metadata for logging @@ -915,7 +944,7 @@ class MLFlowRepository(): } } - def predict(self, model_name: str, data: pd.DataFrame, model_retention: int, + def predict(self, model_name: str, data: pd.DataFrame, model_config: dict, metadata: dict): """ Generate predictions using a cached prediction model. @@ -992,7 +1021,7 @@ class MLFlowRepository(): } def retrain_model(self, data: pd.DataFrame, model_name: str, - model_config: dict) -> tuple: + model_config: dict, metadata: dict) -> tuple: """ Orchestrate the complete model retraining workflow. @@ -1035,15 +1064,41 @@ class MLFlowRepository(): ValueError: If experiment cannot be created or models cannot be loaded Exception: Any other exception during the retraining process """ + + self.logger.debug( + f"Data received for model retraining: {data.to_csv()}", metadata) + transform_flavor = model_config.get('transform_flavor', 'sklearn') predict_flavor = model_config.get('predict_flavor', 'pyfunc') compressed = model_config.get('is_compressed', False) + target_name = model_config.get('target', None) - prediction_model, data_model, experiment = self.create_model_experiment( - model_name, data, transform_flavor, predict_flavor, compressed) - retrain_result = self.perform_model_retrain( - prediction_model, data_model, experiment, model_name, data) - return retrain_result + fit_config = { + 'split_fit_data': model_config.get('split_fit_data', False), + 'split_fit_first': model_config.get('split_fit_first', 'x').lower(), + 'y_type': model_config.get('y_type', 'series').lower() + } + + try: + + prediction_model, data_model, experiment = self.create_model_experiment( + model_name, data, transform_flavor, predict_flavor, compressed, + fit_config, target_name) + experiment = self.perform_model_retrain( + prediction_model, data_model, experiment, model_name, data) + + return { + 'success': True, + 'experiment': experiment, + 'message': 'Model retrained successfully.' + } + except Exception as e: + return { + 'success': False, + 'experiment': None, + 'message': f'Error retraining model {model_name}: {e}', + 'traceback': traceback.format_exc() + } def update_production_model(self, experiment: str, model_name: str) -> dict: """ diff --git a/laborious/workflows/minimal_retrain.py b/laborious/workflows/minimal_retrain.py index 1893e25..98c1e16 100644 --- a/laborious/workflows/minimal_retrain.py +++ b/laborious/workflows/minimal_retrain.py @@ -91,13 +91,29 @@ class MinimalRetrain(): start_to_close_timeout=timedelta(seconds=60) ) + if experiment_response['success']: + + update_report = await workflow.execute_activity_method( + Activities.update_production_model, + { + **metadata, + 'model_name': model_name, + **experiment_response + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + else: + update_report = {} + report = await workflow.execute_activity_method( - Activities.update_production_model, + Activities.format_retrain_report, { **metadata, - 'model_name': model_name, + 'experiment_response': experiment_response, 'model_id': input_data['model_id'], - **experiment_response + 'model_name': model_name, + 'update_report': update_report }, retry_policy=retry_policy, start_to_close_timeout=timedelta(seconds=60) diff --git a/values.yaml b/values.yaml index 4eb83d7..c441f65 100644 --- a/values.yaml +++ b/values.yaml @@ -151,7 +151,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-laborious_temporal.git" - name: GITHUB_BRANCH - value: SIENTIAPDE-1222-ajustar-a-library-para-fazer-o-download-do-courier + value: SIENTIAPDE-1231-ajustar-o-retreino-do-courier-no-laborious - name: PYTHON_APP value: "laborious.worker.worker"