From 7512963e19799761eab255bad3db4cd1e8af5599 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 7 Oct 2025 16:56:16 -0300 Subject: [PATCH] SIENTIAPDE-1231 Enhance MinIO integration and update environment configurations - Added MinIO configuration parameters to .env.example and values.yaml for improved storage management. - Updated requirements.txt to include necessary libraries for MinIO support. - Refactored Activities class to utilize Storage for MinIO interactions. - Enhanced MLFlow class to integrate MinIO for data retrieval during model retraining. - Introduced build_minio_config function to streamline MinIO configuration setup. - Updated minimal_retrain workflow to support data storage in MinIO. --- .env.example | 8 +- laborious/activities/activities.py | 27 ++-- laborious/activities/mlflow.py | 58 ++++++- laborious/activities/storage.py | 142 ++++++++++++++++++ laborious/utils/connectors_config.py | 23 +++ .../utils/repository/minio_repository.py | 115 ++++++++++++++ .../utils/repository/model_repository.py | 99 ++++++++---- laborious/worker/worker.py | 4 + laborious/workflows/minimal_retrain.py | 13 +- requirements.txt | 4 + values.yaml | 11 ++ 11 files changed, 454 insertions(+), 50 deletions(-) create mode 100644 laborious/activities/storage.py create mode 100644 laborious/utils/repository/minio_repository.py diff --git a/.env.example b/.env.example index 865a73b..6e621e9 100644 --- a/.env.example +++ b/.env.example @@ -26,4 +26,10 @@ MONGODB_USERNAME="mongo_user" MONGODB_PASSWORD="mongo_db_password" MONGODB_URL="my-release-mongodb.mongodb.svc.cluster.local:27017" MONGODB_DATABASE="sientia" -MONGODB_TTL_INDEX_HOURS="1" \ No newline at end of file +MONGODB_TTL_INDEX_HOURS="1" + +MINIO_ENDPOINT_URL="http://localhost:9000" +MINIO_ACCESS_KEY="sientia" +MINIO_SECRET_KEY="sientia" +MINIO_REGION_NAME="sa-east-1" +MINIO_DEFAULT_BUCKET="sientia" \ No newline at end of file diff --git a/laborious/activities/activities.py b/laborious/activities/activities.py index ec5ae46..d028064 100644 --- a/laborious/activities/activities.py +++ b/laborious/activities/activities.py @@ -1,7 +1,7 @@ from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): - from sientia_do.temporal.activities.postgres import Postgres + from laborious.activities.storage import Storage from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.observability.logger import Logger from laborious.activities.mlflow import MLFlow @@ -10,7 +10,7 @@ with workflow.unsafe.imports_passed_through(): from typing import Any -class Activities(Postgres, MLFlow, Gates, OPC): +class Activities(Storage, MLFlow, Gates, OPC): """ Main activities orchestrator for the Laborious system. @@ -35,6 +35,7 @@ class Activities(Postgres, MLFlow, Gates, OPC): def __init__(self, postgres_config: dict[str, Any], mlflow_config: dict[str, Any], + minio_config: dict[str, Any], opc_config: dict[str, Any], logger: Logger, notification_handler: NotificationHandler): @@ -58,20 +59,22 @@ class Activities(Postgres, MLFlow, Gates, OPC): Exception: If any parent class initialization fails """ # Initialize parent classes - Postgres.__init__(self, host=postgres_config['host'], - port=postgres_config['port'], - user=postgres_config['user'], - password=postgres_config['password'], - dbname=postgres_config['dbname'], - min_connections=postgres_config['min_connections'], - max_connections=postgres_config['max_connections'], - logger=logger, - notification_handler=notification_handler) + Storage.__init__(self, host=postgres_config['host'], + port=postgres_config['port'], + user=postgres_config['user'], + password=postgres_config['password'], + dbname=postgres_config['dbname'], + min_connections=postgres_config['min_connections'], + max_connections=postgres_config['max_connections'], + minio_config=minio_config, + logger=logger, + notification_handler=notification_handler) MLFlow.__init__(self, mlflow_host=mlflow_config['host'], mlflow_port=mlflow_config['port'], mlflow_username=mlflow_config['username'], mlflow_password=mlflow_config['password'], + minio_config=minio_config, logger=logger, notification_handler=notification_handler) @@ -95,5 +98,5 @@ class Activities(Postgres, MLFlow, Gates, OPC): The method should be called before the application terminates to ensure proper resource cleanup and prevent resource leaks. """ - Postgres.close(self) + Storage.close(self) await OPC.shutdown(self) diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index 87f6ea3..a09c297 100644 --- a/laborious/activities/mlflow.py +++ b/laborious/activities/mlflow.py @@ -1,9 +1,9 @@ +from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): - from datetime import datetime - from pandas import Timestamp, to_datetime + from pandas import to_datetime from sientia_do.temporal.constants import DATETIME_FORMAT, DATETIME_FORMAT_WITH_TZ from sientia_do.temporal.activities.base import BaseActivity from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler @@ -15,6 +15,7 @@ with workflow.unsafe.imports_passed_through(): import numpy as np from pandas import DataFrame import traceback + from laborious.utils.repository.minio_repository import MinioRepository class MLFlow(BaseActivity): @@ -37,7 +38,8 @@ class MLFlow(BaseActivity): """ def __init__(self, mlflow_host: str, mlflow_port: int, mlflow_username: str, - mlflow_password: str, logger: Logger, notification_handler: NotificationHandler): + minio_config: dict[str, Any], mlflow_password: str, + logger: Logger, notification_handler: NotificationHandler): """ Initialize MLFlow activities with server configuration. @@ -63,6 +65,26 @@ class MLFlow(BaseActivity): f"{mlflow_host}:{mlflow_port}", mlflow_username, mlflow_password, logger ) + if not hasattr(self, 'minio_repository'): + self.minio_repository = MinioRepository( + logger=logger, + notification_handler=notification_handler, + minio_endpoint_url=minio_config['endpoint_url'], + minio_access_key=minio_config['access_key'], + minio_secret_key=minio_config['secret_key'], + minio_region_name=minio_config['region_name'], + minio_default_bucket=minio_config['default_bucket']) + + if self.minio_repository is None: + self.minio_repository = MinioRepository( + logger=logger, + notification_handler=notification_handler, + minio_endpoint_url=minio_config['endpoint_url'], + minio_access_key=minio_config['access_key'], + minio_secret_key=minio_config['secret_key'], + minio_region_name=minio_config['region_name'], + minio_default_bucket=minio_config['default_bucket']) + @activity.defn(name="request_transform") async def request_transform(self, input_data: dict[str, Any]) -> dict[str, Any]: """ @@ -223,7 +245,35 @@ class MLFlow(BaseActivity): Exception: If retraining fails or encounters critical errors """ metadata = input_data['metadata'] - data = DataFrame(input_data['data']) + object_key = input_data['object_key'] + + self.info(f'Loading retrain data from Key: {object_key}', metadata) + + try: + + data = self.minio_repository.get_parquet_as_dataframe( + object_key=object_key, metadata=metadata) + except Exception as e: + trace = traceback.format_exc() + self.send_notification( + metadata=metadata, + notification_id='ERROR_LOADING_RETRAIN_DATA', + message=f'Error loading retrain data: {e}', + block='retrain_model', + level=NotificationLevel.ERROR, + attachment_content=trace + ) + self.error(trace, metadata) + return { + 'success': False, + 'message': f'Error loading retrain data: {e}', + 'traceback': trace, + 'timestamp': now().strftime(DATETIME_FORMAT_MS_WITH_TZ) + } + + self.debug( + f'Retrain data loaded successfully: shape {data.shape}', metadata) + model_name = input_data['model_name'] model_config = input_data.get('model_config', {}) diff --git a/laborious/activities/storage.py b/laborious/activities/storage.py new file mode 100644 index 0000000..ef61fb4 --- /dev/null +++ b/laborious/activities/storage.py @@ -0,0 +1,142 @@ +from temporalio import activity, workflow + +from laborious.utils.repository.minio_repository import MinioRepository + + +with workflow.unsafe.imports_passed_through(): + # Extend the Temporal Postgres activities for convenient query -> MinIO export + from sientia_do.temporal.activities.postgres import Postgres + from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler + from sientia_do.observability.logger import Logger + from sientia_do.temporal.constants import now + from sientia_do.notifications.models import NotificationLevel + from typing import Any + import traceback + import pandas as pd + +DATETIME_FILENAME_FORMAT = "%Y-%m-%d_%H-%M-%S" + + +class Storage(Postgres): + """ + Extensions for Postgres activities with a helper to export query results + directly to MinIO as Parquet and return the object name. + """ + + def __init__(self, + host: str, + port: int, + user: str, + password: str, + dbname: str, + min_connections: int, + max_connections: int, + minio_config: dict[str, Any], + logger: Logger, + notification_handler: NotificationHandler): + super().__init__(host=host, + port=port, + user=user, + password=password, + dbname=dbname, + min_connections=min_connections, + max_connections=max_connections, + logger=logger, + notification_handler=notification_handler) + + if not hasattr(self, 'minio_repository'): + self.minio_repository = MinioRepository( + logger=logger, + notification_handler=notification_handler, + minio_endpoint_url=minio_config['endpoint_url'], + minio_access_key=minio_config['access_key'], + minio_secret_key=minio_config['secret_key'], + minio_region_name=minio_config['region_name'], + minio_default_bucket=minio_config['default_bucket']) + + if self.minio_repository is None: + self.minio_repository = MinioRepository( + logger=logger, + notification_handler=notification_handler, + minio_endpoint_url=minio_config['endpoint_url'], + minio_access_key=minio_config['access_key'], + minio_secret_key=minio_config['secret_key'], + minio_region_name=minio_config['region_name'], + minio_default_bucket=minio_config['default_bucket']) + + @activity.defn(name='query_to_minio') + async def query_to_minio(self, input_data: dict[str, Any]) -> dict[str, Any]: + """ + Execute SQL query, write result as Parquet to MinIO, and return object name. + + Args (input_data): + metadata (dict): Workflow metadata + query (str): SQL query + model_name (str): Model name for object naming + object_prefix (str, optional): Prefix inside bucket (default: datasets/retrain) + + Returns: + dict: { success: bool, object_name: str, uri: str } + """ + + metadata = input_data.get('metadata', {}) + object_prefix = input_data.get('object_prefix', 'datasets/retrain') + + timestamp = now().strftime(DATETIME_FILENAME_FORMAT) + object_name = f"{object_prefix}_{timestamp}.parquet" + uri = f"s3://{self.minio_repository.minio_bucket}/{object_name}" + + try: + data = await self.load_custom_query(input_data) + if not data: + self.error( + f"query_to_minio failed: No data returned from query", metadata) + return {"success": False, "message": "No data returned from query"} + + # Ensure we have a DataFrame + data = pd.DataFrame(data) + + # Write parquet to memory and upload via persistent client + self.minio_repository.store_dataframe_as_parquet( + dataframe=data, + uri=uri, + object_name=object_name, + metadata=metadata + ) + + return {"success": True, "object_key": object_name, "uri": uri} + except Exception as e: + trace = traceback.format_exc() + self.send_notification( + metadata=metadata, + notification_id="ERROR_LOADING_CUSTOM_QUERY", + message=f"Error fetching data from query: {e}", + block="load_custom_query", + level=NotificationLevel.ERROR, + attachment_content=trace + ) + + self.error(trace, metadata) + + return {"success": False, "message": str(e)} + + def close(self) -> None: + """Close Storage resources (MinIO client and Postgres engine).""" + try: + if hasattr(self, 's3_client') and self.s3_client is not None: + try: + self.s3_client.close() + finally: + self.s3_client = None + finally: + # Ensure Postgres resources are disposed as well + try: + super().close() + except Exception: + pass + + def __del__(self): + try: + self.close() + except Exception: + pass diff --git a/laborious/utils/connectors_config.py b/laborious/utils/connectors_config.py index 41c933f..969b95c 100644 --- a/laborious/utils/connectors_config.py +++ b/laborious/utils/connectors_config.py @@ -127,3 +127,26 @@ def build_mongodb_config() -> Dict[str, Any]: 'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia'), 'ttl_index_seconds': int(getenv('MONGODB_TTL_INDEX_HOURS', '1')) * 3600 } + + +def build_minio_config() -> Dict[str, Any]: + """ + Build MinIO (S3-compatible) configuration from environment variables. + + Environment Variables: + MINIO_ENDPOINT: MinIO endpoint including scheme (default: http://localhost:9000) + MINIO_ACCESS_KEY: Access key (default: minioadmin) + MINIO_SECRET_KEY: Secret key (default: minioadmin) + MINIO_REGION: Region name for S3 client (default: us-east-1) + MINIO_BUCKET_DEFAULT: Default bucket for uploads (default: laborious) + + Returns: + dict: MinIO configuration dictionary + """ + return { + 'endpoint_url': getenv('MINIO_ENDPOINT_URL', 'http://localhost:9000'), + 'access_key': getenv('MINIO_ACCESS_KEY', 'minioadmin'), + 'secret_key': getenv('MINIO_SECRET_KEY', 'minioadmin'), + 'region_name': getenv('MINIO_REGION_NAME', 'us-east-1'), + 'default_bucket': getenv('MINIO_DEFAULT_BUCKET', 'laborious') + } diff --git a/laborious/utils/repository/minio_repository.py b/laborious/utils/repository/minio_repository.py new file mode 100644 index 0000000..d957e65 --- /dev/null +++ b/laborious/utils/repository/minio_repository.py @@ -0,0 +1,115 @@ +from io import BytesIO +import traceback +import boto3 +from botocore.config import Config +from pandas import DataFrame, read_parquet +from sientia_do.observability.logger import Logger +from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler +from sientia_do.notifications.models import NotificationLevel +from typing import Any +from botocore.exceptions import ClientError + + +class MinioRepository(): + def __init__(self, + minio_endpoint_url: str, + minio_access_key: str, + minio_secret_key: str, + minio_region_name: str, + minio_default_bucket: str, + logger: Logger, + notification_handler: NotificationHandler): + + # MinIO settings shared with pandas s3fs + self.storage_options = { + 'key': minio_access_key, + 'secret': minio_secret_key, + 'client_kwargs': {'endpoint_url': minio_endpoint_url} + } + self.minio_bucket = minio_default_bucket + self.minio_endpoint_url = minio_endpoint_url + self.minio_region_name = minio_region_name + + logger.info( + f"Connecting to MinIO at {self.minio_endpoint_url}, default bucket: {self.minio_bucket}") + + # Reusable MinIO client + self.s3_client = boto3.client( + 's3', + endpoint_url=self.minio_endpoint_url, + aws_access_key_id=self.storage_options['key'], + aws_secret_access_key=self.storage_options['secret'], + region_name=self.minio_region_name, + config=Config( + signature_version='s3v4', + s3={'addressing_style': 'path'}, + retries={'max_attempts': 5, 'mode': 'standard'}, + connect_timeout=5, + read_timeout=120, + ), + ) + self._bucket_checked = False + + self.logger = logger + self.notification_handler = notification_handler + + def ensure_bucket_exists(self, metadata: dict[str, Any]) -> bool: + """ + Ensure the MinIO bucket exists; create it if necessary. + """ + if self._bucket_checked: + return True + + try: + self.logger.custom_info( + f"Checking if bucket '{self.minio_bucket}' exists", metadata) + self.s3_client.head_bucket(Bucket=self.minio_bucket) + self._bucket_checked = True + return True + except ClientError: + try: + self.logger.custom_info( + f"Creating bucket '{self.minio_bucket}'", metadata) + self.s3_client.create_bucket(Bucket=self.minio_bucket) + self._bucket_checked = True + return True + except ClientError as ce: + trace = traceback.format_exc() + self.notification_handler.send_notification( + metadata=metadata, + notification_id="ERROR_CREATING_MINIO_BUCKET", + message=f"Failed to ensure bucket '{self.minio_bucket}': {ce}", + block="ensure_bucket_exists", + level=NotificationLevel.ERROR, + attachment_content=str(ce) + ) + self.logger.custom_error(trace, metadata) + return False + + def store_dataframe_as_parquet(self, dataframe: DataFrame, uri: str, + object_name: str, metadata: dict[str, Any]): + + self.ensure_bucket_exists(metadata) + + self.logger.custom_info( + f"Storing dataframe as parquet in {uri}", metadata) + + buffer = BytesIO() + dataframe.to_parquet(buffer, engine='pyarrow', index=True) + buffer.seek(0) + self.s3_client.put_object( + Bucket=self.minio_bucket, Key=object_name, Body=buffer.getvalue()) + + self.logger.custom_info( + f"Dataframe stored as parquet in {uri}", metadata) + + def get_parquet_as_dataframe(self, object_key: str, metadata: dict[str, Any]) -> DataFrame: + self.logger.custom_info( + f"Getting parquet as dataframe from {object_key}", metadata) + + response = self.s3_client.get_object( + Bucket=self.minio_bucket, Key=object_key) + + # Read the content into a BytesIO buffer to support seek operations + buffer = BytesIO(response['Body'].read()) + return read_parquet(buffer) diff --git a/laborious/utils/repository/model_repository.py b/laborious/utils/repository/model_repository.py index 54b1985..b194a69 100644 --- a/laborious/utils/repository/model_repository.py +++ b/laborious/utils/repository/model_repository.py @@ -244,7 +244,7 @@ class MLFlowRepository(): output_dir ) - def load_predict_model(self, model_name: str, flavor: str = 'pyfunc', + def load_predict_model(self, model_name: str, flavor: str = 'sklearn', artifact_path: str | None = None) -> Any: """ Downloads a predictive model from the MLflow Model Registry. @@ -386,7 +386,7 @@ class MLFlowRepository(): f"Could not load model from {pickle_path} - unknown or corrupted format") def download_model(self, model_name: str, model_type: str, flavor: str, - download_artifacts: bool = False) -> Any: + download_artifacts: bool = False) -> tuple[Any, str]: """ Download model based on type (predict or transform). @@ -397,7 +397,7 @@ class MLFlowRepository(): download_artifacts (bool): Whether to download artifacts Returns: - Any: Model object + tuple[Any, str]: Model object and artifact path if model is compressed """ self.logger.info( @@ -422,7 +422,7 @@ class MLFlowRepository(): model = self.load_transform_model( model_name, flavor, artifact_path) - return model + return model, artifact_path """ Functions related to data format @@ -633,9 +633,9 @@ class MLFlowRepository(): """ def create_model_experiment(self, model_name: str, data: pd.DataFrame, - transform_flavor: str = 'sklearn', predict_flavor: str = 'pyfunc', + transform_config: dict = {}, predict_config: dict = {}, fit_config: dict = {}, target_name: str = None, - metadata: dict = {}) -> tuple: + is_compressed: bool = False, metadata: dict = {}) -> tuple: """ Create a new MLFlow experiment for model retraining. @@ -649,10 +649,11 @@ class MLFlowRepository(): Args: model_name (str): Name of the MLFlow model to retrain data (pd.DataFrame): Training data for model retraining - transform_flavor (str): Flavor for transformation model - predict_flavor (str): Flavor for prediction model + transform_config (dict): Configuration for transformation model + predict_config (dict): Configuration for prediction model fit_config (dict): Fit configuration target_name (str): Target name + is_compressed (bool): Whether model is compressed metadata (dict): Metadata for logging Returns: @@ -664,7 +665,7 @@ class MLFlowRepository(): self.logger.custom_info( f"Starting model experiment creation for {model_name}", metadata) self.logger.custom_debug( - f"Model configuration - transform_flavor: {transform_flavor}, predict_flavor: {predict_flavor}, fit_config: {fit_config}, target_name: {target_name}", metadata) + f"Model configuration - transform_config: {transform_config}, predict_config: {predict_config}, fit_config: {fit_config}, target_name: {target_name}", metadata) latest_production_id = self.get_model_run_id( model_name, stage="Production" @@ -675,17 +676,24 @@ class MLFlowRepository(): self.logger.custom_info( f"Loading transformation model for {model_name}", metadata) - data_model = self.download_model( + transform_flavor = transform_config.get('flavor', 'sklearn') + transformed_is_compressed = transform_config.get( + 'is_compressed', False) + + data_model, data_artifact_path = self.download_model( model_name=model_name, model_type="transform", flavor=transform_flavor, - download_artifacts=(transform_flavor == 'pyfunc') + download_artifacts=transformed_is_compressed ) self.logger.custom_info( f"Loading prediction model for {model_name}", metadata) - prediction_model = self.download_model( + predict_flavor = predict_config.get('flavor', 'pyfunc') + predicted_is_compressed = predict_config.get('is_compressed', False) + + prediction_model, prediction_artifact_path = self.download_model( model_name=model_name, model_type="predict", flavor=predict_flavor, - download_artifacts=(predict_flavor == 'pyfunc') + download_artifacts=predicted_is_compressed ) self.logger.custom_debug( @@ -784,14 +792,36 @@ class MLFlowRepository(): self.logger.custom_info( f"Model experiment creation completed successfully for {model_name}", metadata) - return prediction_model, data_model, experiment + + retrain_data = { + 'prediction_model': prediction_model, + 'prediction_artifact_path': prediction_artifact_path, + 'data_model': data_model, + 'data_artifact_path': data_artifact_path, + 'experiment': experiment + } + return retrain_data + + def log_model(self, model: Any, artifact_local_path: str, flavor: str, model_type: str): + if artifact_local_path: + mlflow.log_artifact(artifact_local_path, artifact_path="") + else: + if flavor == 'sklearn': + mlflow.sklearn.log_model(model, model_type) + elif flavor == 'pyfunc': + mlflow.pyfunc.log_model(model, model_type) + elif flavor == 'pytorch': + mlflow.pytorch.log_model(model, model_type) + else: + raise ValueError( + "Invalid flavor. Use 'sklearn' or 'pyfunc' or 'pytorch'.") def perform_model_retrain(self, - prediction_model, - data_model, - experiment: str, model_name: str, data: pd.DataFrame, + retrain_data: dict, + transform_config: dict = {}, + predict_config: dict = {}, metadata: dict = {}): """ Execute the complete model retraining process in MLFlow. @@ -809,6 +839,7 @@ class MLFlowRepository(): experiment (str): MLFlow experiment name for the retraining model_name (str): Name of the model being retrained data (pd.DataFrame): Training data used for retraining + is_compressed (bool): Whether model is compressed metadata (dict): Metadata for logging Returns: @@ -816,9 +847,19 @@ class MLFlowRepository(): - status_message (str): Success confirmation message - experiment_name (str): Name of the experiment """ + + prediction_model = retrain_data['prediction_model'] + data_model = retrain_data['data_model'] + experiment = retrain_data['experiment'] + prediction_artifact_path = retrain_data['prediction_artifact_path'] + data_artifact_path = retrain_data['data_artifact_path'] + self.logger.custom_info( f"Starting model retraining process for {model_name} in experiment {experiment}", metadata) + transform_flavor = transform_config.get('flavor', 'sklearn') + predict_flavor = predict_config.get('flavor', 'sklearn') + pred_model_atributes = vars(prediction_model) # load class attributes data_model_atributes = vars(data_model) # load class attributes experiment_description = f"Retrain model {model_name} with new data" @@ -845,7 +886,8 @@ class MLFlowRepository(): mlflow.log_param(name_atribute, val_atribute) # dynamic parameters, including model itself - mlflow.sklearn.log_model(data_model, "data_model") + self.log_model(data_model, data_artifact_path, + transform_flavor, "data_model") makedirs("tmp/retrain_data", exist_ok=True) @@ -856,7 +898,8 @@ class MLFlowRepository(): mlflow.log_artifact(file_path) # dynamic parameters, including model itself - mlflow.sklearn.log_model(prediction_model, "prediction_model") + self.log_model(prediction_model, prediction_artifact_path, + predict_flavor, "prediction_model") mlflow.log_param("retrain", True) # clear temp file @@ -1044,7 +1087,7 @@ class MLFlowRepository(): """ model_retention = model_config.get('retention_minutes', 0) - flavor = model_config.get('predict_flavor', 'pyfunc') + flavor = model_config.get('predict_flavor', 'sklearn') try: @@ -1147,8 +1190,8 @@ class MLFlowRepository(): target_name = model_config.get('target', None) - transform_flavor = model_config.get('transform_flavor', 'sklearn') - predict_flavor = model_config.get('predict_flavor', 'pyfunc') + transform_config = model_config.get('transform_config', {}) + predict_config = model_config.get('predict_config', {}) fit_config = { 'split_fit_data': model_config.get('split_fit_data', False), @@ -1157,21 +1200,19 @@ class MLFlowRepository(): } self.logger.custom_debug( - f"Model configuration - transform_flavor: {transform_flavor}, predict_flavor: {predict_flavor}, fit_config: {fit_config}, target_name: {target_name}", metadata) + f"Model configuration - transform_config: {transform_config}, predict_config: {predict_config}, fit_config: {fit_config}, target_name: {target_name}", metadata) try: self.logger.custom_info( "Creating model experiment environment", metadata) - prediction_model, data_model, experiment = self.create_model_experiment( - model_name=model_name, data=data, transform_flavor=transform_flavor, - predict_flavor=predict_flavor, fit_config=fit_config, target_name=target_name, - metadata=metadata) + retrain_data = self.create_model_experiment( + model_name=model_name, data=data, transform_config=transform_config, predict_config=predict_config, fit_config=fit_config, target_name=target_name, metadata=metadata) self.logger.custom_info( - f"Model experiment created successfully: {experiment}", metadata) + f"Model experiment created successfully: {retrain_data}", metadata) self.logger.custom_info("Saving model retrain", metadata) experiment = self.perform_model_retrain( - prediction_model, data_model, experiment, model_name, data, metadata) + model_name, data, retrain_data, transform_config, predict_config, metadata) self.logger.custom_info( f"Model retraining completed successfully for experiment: {experiment}", metadata) diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index ad2e6c0..bb74fb3 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -44,6 +44,7 @@ with workflow.unsafe.imports_passed_through(): from laborious.utils.connectors_config import ( build_postgres_config, build_mlflow_config, + build_minio_config, build_opc_config, build_mongodb_config ) @@ -152,6 +153,7 @@ async def main(): activities = Activities( postgres_config=build_postgres_config(), mlflow_config=build_mlflow_config(), + minio_config=build_minio_config(), opc_config=build_opc_config(), logger=logger, notification_handler=notification_handler @@ -190,6 +192,7 @@ async def main(): workflows=[MinimalRetrain], activities=[ activities.load_custom_query, + activities.query_to_minio, activities.retrain_model, activities.update_production_model, activities.format_retrain_report, @@ -211,6 +214,7 @@ async def main(): # MLFlow activities.request_predict, activities.request_transform, + activities.query_to_minio, # Gates activities.input_gate, activities.mlflow_response_gate, diff --git a/laborious/workflows/minimal_retrain.py b/laborious/workflows/minimal_retrain.py index 607f3a5..ebdb0b2 100644 --- a/laborious/workflows/minimal_retrain.py +++ b/laborious/workflows/minimal_retrain.py @@ -70,22 +70,27 @@ class MinimalRetrain(): model_name = input_data['model_name'] model_config = input_data.get('model_config', {}) - data = await workflow.execute_local_activity_method( - Activities.load_custom_query, + storage_result = await workflow.execute_local_activity_method( + Activities.query_to_minio, { **metadata, 'query': input_data['query'], - 'datetime_columns': input_data.get('datetime_columns', []) + 'datetime_columns': input_data.get('datetime_columns', []), + 'model_name': model_name, + 'object_prefix': f'retrain_datasets/{model_name}/data' }, retry_policy=retry_policy, start_to_close_timeout=timedelta(seconds=600) ) + if not storage_result['success']: + return + experiment_response = await workflow.execute_activity_method( Activities.retrain_model, { **metadata, - 'data': data, + 'object_key': storage_result['object_key'], 'model_name': model_name, 'model_config': model_config }, diff --git a/requirements.txt b/requirements.txt index 1ce9f9c..3ba54e8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,3 +6,7 @@ redis git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.6 git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.39.0 prometheus-client +botocore +boto3 +s3fs +pyarrow diff --git a/values.yaml b/values.yaml index c441f65..3f2b9f8 100644 --- a/values.yaml +++ b/values.yaml @@ -213,6 +213,17 @@ env: - name: MONGODB_TTL_INDEX_HOURS value: "1" + - name: MINIO_ENDPOINT_URL + value: "http://sientia-minio-minio.sientia.svc.cluster.local:9000" + - name: MINIO_ACCESS_KEY + value: "admin" + - name: MINIO_SECRET_KEY + value: "FvcxOPX55j" + - name: MINIO_REGION_NAME + value: "sa-east-1" + - name: MINIO_DEFAULT_BUCKET + value: "sientia" + ssh: enabled: true secretName: git-ssh-key-sientia-laborious-worker