From d43f08d272c4ca2a24f9830725897957384abd84 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 20 Mar 2026 09:22:08 -0300 Subject: [PATCH] SIENTIAPDE-1712 Enhance README and Implement Drift Detection and Metrics Workflows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Added new sections in README for Drift Workflow and Simple Metrics Workflow, detailing their execution flows and functionalities. - Introduced `drift.py` for data drift detection, comparing current data against reference datasets. - Added `simple_metrics.py` for calculating regression metrics (RMSE, MSE, MAE, R²). - Updated `values.yaml` to include configuration for MinIO retention hours and offload threshold. - Refactored `minio_dataframe_payload.py` to use the new offload threshold environment variable. - Adjusted tests to reflect changes in environment variable handling for MinIO offload threshold. --- README.md | 217 ++++++++++++++---- .../utils/models/minio_dataframe_payload.py | 2 +- tests/conftest.py | 4 +- values.yaml | 4 + 4 files changed, 176 insertions(+), 51 deletions(-) diff --git a/README.md b/README.md index a02f756..c9c37d6 100644 --- a/README.md +++ b/README.md @@ -19,6 +19,8 @@ A comprehensive, Temporal-based ML orchestration system for industrial data proc - [Prediction Process Workflow](#2-prediction-process-workflow-prediction_processpy) - [Format and Export Prediction Workflow](#3-format-and-export-prediction-workflow-format_and_export_predictionpy) - [Minimal Retrain Workflow](#4-minimal-retrain-workflow-minimal_retrainpy) + - [Drift Workflow](#5-drift-workflow-driftpy) + - [Simple Metrics Workflow](#6-simple-metrics-workflow-simple_metricspy) - [Installation & Setup](#installation--setup) - [Prerequisites](#prerequisites) - [Environment Setup](#environment-setup) @@ -77,13 +79,16 @@ A comprehensive, Temporal-based ML orchestration system for industrial data proc ### Advanced Capabilities - **Incremental Data Processing**: Timestamp-based loading to avoid reprocessing - **Configurable Data Retention**: Model retention policies with automatic cleanup +- **MinIO Payload Offload**: Automatic offload of large DataFrames to MinIO with retention cleanup +- **Data Drift Detection**: Univariate and multivariate drift monitoring against reference data +- **Regression Metrics**: Automated RMSE, MSE, MAE, R² calculation and export - **Notification System**: Integrated alerting via MongoDB - **Scalable Architecture**: Kubernetes-ready with horizontal scaling - **Model Retraining**: Automated retraining workflows with production model updates ### Development & Quality Assurance - **Code Quality Tools**: Ruff (lint/format), mypy (types), Bandit (security) -- **Automated Validation**: `validate.sh` and CI quality gates +- **Automated Validation**: CI quality gates and individual tool commands - **Comprehensive Testing**: pytest with async support and high coverage - **Type Safety**: Static type checking with mypy - **Coverage Visualization**: Coverage Gutters integration @@ -129,6 +134,8 @@ Laborious uses a Temporal-based architecture with strong separation of concerns - `sub_workflows/prediction_process.py`: Core prediction pipeline - `sub_workflows/format_and_export_prediction.py`: Formatting and export - `minimal_retrain.py`: Automated model retraining and production update +- `drift.py`: Data drift detection and monitoring +- `simple_metrics.py`: Regression metrics calculation (RMSE, MSE, MAE, R²) #### **Activities (`laborious/activities/`)** - `gates.py`: Data quality validation, filtering, and data formatting operations @@ -139,16 +146,26 @@ Laborious uses a Temporal-based architecture with strong separation of concerns - MLFlow model transformation and prediction - Model retraining and production updates - Reference data retrieval from MLflow Model Registry +- `storage.py`: PostgreSQL queries and MinIO-aware data loading + - `load_query_with_minio_offload`: SQL load with automatic MinIO offload + - `export_payload_to_postgres`: Resolve MinIO payloads and export to Postgres + - `cleanup_minio_objects_expired`: Retention-based MinIO object cleanup + - `query_to_minio`: Legacy parquet upload for retraining data +- `model_metrics.py`: Drift detection and regression metrics + - Univariate and multivariate drift calculation + - Simple metrics (RMSE, MSE, MAE, R²) - `opc.py`: OPC UA export to industrial systems (optional) - `api.py`: PI Web API export operations (optional) - Prediction and confidence data writing to PI Web API - Error handling and notification integration -- `activities.py`: Aggregates activity interfaces +- `activities.py`: Aggregates all activity interfaces (Storage, MLFlow, Gates, OPC, ModelMetrics, API) #### **Data Services (`laborious/utils/`)** - `connectors_config.py`: Env-driven configuration builders +- `models/minio_dataframe_payload.py`: MinIO-offloaded DataFrame payload model - `repository/model_repository.py`: MLFlow operations and retraining - `repository/opc_repository.py`: OPC communication and writes +- `repository/minio_manager.py`: MinIO object storage operations - `filters/conditional_filters.py` and `filters/mlflow_filters.py` ### Data Flow Architecture @@ -167,6 +184,18 @@ Training Data → Model Retraining → Quality Validation → Production Update → Notification & Monitoring ``` +#### **3. Drift Detection Pipeline** +``` +Target Data (PostgreSQL) + Reference Data (MLFlow) → +Drift Calculation (univariate + multivariate) → PostgreSQL Export +``` + +#### **4. Simple Metrics Pipeline** +``` +Predictions + Targets (PostgreSQL JOIN) → +Metrics Calculation (RMSE, MSE, MAE, R²) → PostgreSQL Export +``` + ### Security Architecture #### **Authentication & Authorization** @@ -262,20 +291,21 @@ The **PredictionProcess** workflow implements the core prediction pipeline for M - **Prediction Export**: Delegates prediction formatting and export operations #### Execution Flow -1. **Timestamp Retrieval**: Gets the last processed timestamp for incremental processing -2. **Input Data Gate**: Applies configured filters for data quality validation -3. **Path Decision**: Determines processing path based on filter results -4. **MLFlow Transform**: Requests data transformation using MLFlow models -5. **Response Validation**: Filters transform responses for quality assurance +1. **Input Data Gate**: Applies configured filters for data quality validation +2. **Path Decision**: Determines processing path based on filter results +3. **MLFlow Transform**: Requests data transformation using MLFlow models +4. **Response Validation**: Filters transform responses for quality assurance +5. **Content Validation**: Filters transformed data content for quality check 6. **MLFlow Prediction**: Executes prediction using transformed data -7. **Content Validation**: Filters prediction responses for final quality check +7. **Prediction Response Validation**: Filters prediction responses for final quality check 8. **Export Delegation**: Delegates to FormatAndExportPrediction workflow +9. **MinIO Cleanup**: Cleans up expired offloaded payloads (if any, in `finally` block) #### Key Features - **Configurable Quality Gates**: Multiple filter types with policy-based configuration - **Flexible Path Handling**: Configurable decision paths (STOP, CONTINUE, REPEAT) - **MLFlow Integration**: Comprehensive model management and inference -- **Incremental Processing**: Timestamp-based data processing optimization +- **MinIO Cleanup**: Automatic retention-based cleanup of offloaded payloads - **Comprehensive Monitoring**: Detailed metrics and error reporting #### Input Parameters @@ -320,12 +350,12 @@ The **PredictionProcess** workflow implements the core prediction pipeline for M #### Architecture Diagram ```mermaid flowchart LR - A[1. get_last_timestamp] --> B[2. input_gate] --> C[3. request_transform] --> D[4. mlflow_response_gate] --> E[5. mlflow_content_gate] --> F[6. request_predict] --> G[7. mlflow_response_gate] --> H[8. format_and_export_prediction🔃] + A[1. input_gate] --> B[2. request_transform] --> C[3. mlflow_response_gate] --> D[4. mlflow_content_gate] --> E[5. request_predict] --> F[6. mlflow_response_gate] --> G[7. format_and_export_prediction🔃] + G --> H[8. cleanup_minio_objects_expired] - A -.-> Redis[(Redis)] - C -.-> MLFlow[MLFlow] - F -.-> MLFlow[MLFlow] - G -.-> Filters[MLFlow Filters] + B -.-> MLFlow[MLFlow] + E -.-> MLFlow[MLFlow] + H -.-> MinIO[(MinIO)] ``` ### 3. Format and Export Prediction Workflow (`format_and_export_prediction.py`) @@ -403,6 +433,76 @@ flowchart LR D -.-> PostgreSQL[(PostgreSQL)] ``` +### 5. Drift Workflow (`drift.py`) + +The **Drift** workflow detects data drift by comparing current data against a reference dataset from the MLflow Model Registry. + +#### Execution Flow +1. **Data Loading**: Loads target data and reference data in parallel +2. **Drift Calculation**: Calculates univariate and multivariate drift metrics +3. **Data Export**: Exports drift metrics to PostgreSQL + +#### Architecture Diagram +```mermaid +flowchart LR + A[1. load_custom_query] --> C[3. calculate_drift] --> D[4. export_data_to_postgres] + B[2. get_reference_data] --> C + + A -.-> Database[(Database)] + B -.-> MLFlow[MLFlow] + D -.-> PostgreSQL[(PostgreSQL)] +``` + +#### Input Parameters +```json +{ + "schedule_name": "hourly_drift", + "model_name": "temperature_model", + "model_id": "temp_001", + "schema": "sientia_data", + "source_table_name": "laborious_data", + "target_table_name": "drift_metrics", + "interval": 60, + "model_config": { "target": "temperature" }, + "drift_metrics": ["kolmogorov_smirnov", "jensen_shannon", "wasserstein"], + "chunk_period": "min" +} +``` + +### 6. Simple Metrics Workflow (`simple_metrics.py`) + +The **SimpleMetrics** workflow calculates regression metrics (RMSE, MSE, MAE, R²) by comparing predictions against actual target values. + +#### Execution Flow +1. **Data Loading**: Loads prediction vs target data via a JOIN query +2. **Metrics Calculation**: Calculates configured regression metrics +3. **Data Export**: Exports metrics to PostgreSQL + +#### Architecture Diagram +```mermaid +flowchart LR + A[1. load_custom_query] --> B[2. calculate_simple_metrics] --> C[3. export_data_to_postgres] + + A -.-> Database[(Database)] + C -.-> PostgreSQL[(PostgreSQL)] +``` + +#### Input Parameters +```json +{ + "schedule_name": "hourly_metrics", + "model_name": "temperature_model", + "model_id": "temp_001", + "schema": "sientia_data", + "predictions_table_name": "predictions", + "data_table_name": "laborious_data", + "target_table_name": "simple_metrics", + "interval_minutes": 60, + "model_config": { "target": "temperature" }, + "metrics": ["rmse", "mse", "mae", "r2"] +} +``` + ## 📋 Prerequisites - Python 3.11+ @@ -527,8 +627,8 @@ pytest pytest --cov=laborious --cov-report=html # Run specific test categories -pytest tests/activities/ -pytest tests/workflow/ +pytest tests/laborious/activities/ +pytest tests/laborious/workflows/ ``` ### Manual Application Execution @@ -569,18 +669,7 @@ pip install -r requirements-dev.txt ### Complete Validation -Option 1 (recommended): -```bash -./validate.sh -``` -The script runs, in order: -1. Format check (Ruff) -2. Linting (Ruff) -3. Type checking (mypy) -4. Security analysis (Bandit) -5. Tests with coverage (pytest) - -Option 2 (individual commands): +Run each validation step individually: ```bash ruff format --check laborious/ tests/ ruff check laborious/ tests/ @@ -606,7 +695,7 @@ The workflow at `.github/workflows/quality-gate.yml` executes validations on eac ### Best Practices -- Run `./validate.sh` before committing +- Run all validation steps before committing - Use `ruff check --watch` for continuous feedback - Add type hints and tests for new code @@ -615,15 +704,34 @@ The workflow at `.github/workflows/quality-gate.yml` executes validations on eac ### Test Structure ``` tests/ -├── activities/ # Activity implementation tests -│ ├── test_gates.py # Data quality gates and formatting tests -│ ├── test_mlflow.py # MLFlow operations and reference data tests -│ └── ... # Other activity tests -├── workflows/ # Workflow orchestration tests -│ └── subworkflows/ # Sub-workflow tests -│ └── test_format_and_export_prediction.py # Export workflow tests -├── utils/ # Utility function tests -└── integration/ # End-to-end workflow tests +├── conftest.py # Global fixtures and env setup +├── laborious/ +│ ├── activities/ # Activity implementation tests +│ │ ├── test_activities.py # Activities aggregator tests +│ │ ├── test_gates.py # Data quality gates and formatting tests +│ │ ├── test_mlflow.py # MLFlow operations and reference data tests +│ │ ├── test_storage.py # Storage and MinIO offload tests +│ │ ├── test_model_metrics.py # Drift and simple metrics tests +│ │ ├── test_opc.py # OPC operations tests +│ │ └── test_api.py # PI Web API operations tests +│ ├── workflows/ # Workflow orchestration tests +│ │ ├── test_predictions_batch.py +│ │ ├── test_minimal_retrain.py +│ │ ├── test_drift.py +│ │ ├── test_simple_metrics.py +│ │ └── subworkflows/ +│ │ ├── test_prediction_process.py +│ │ └── test_format_and_export_prediction.py +│ └── utils/ # Utility function tests +│ ├── test_connectors_config.py +│ ├── models/ +│ │ └── test_minio_dataframe_payload.py +│ ├── filters/ +│ │ ├── test_conditional_filters.py +│ │ └── test_mlflow_filters.py +│ └── repository/ +│ ├── test_model_repository.py +│ └── test_opc_repository.py ``` ### Test Coverage @@ -643,8 +751,8 @@ pip install pytest pytest-cov pytest-asyncio pytest --cov=laborious --cov-report=html # Run specific test modules -pytest tests/activities/test_gates.py -pytest tests/workflow/test_predictions_batch.py +pytest tests/laborious/activities/test_gates.py +pytest tests/laborious/workflows/test_predictions_batch.py ``` ## 📊 Monitoring and Metrics @@ -716,8 +824,13 @@ The Laborious system exposes comprehensive Prometheus metrics for operational vi | `HTTP_METRICS_PORT` | Prometheus metrics port | `9090` | No | | `HTTP_SDK_METRICS_PORT` | Temporal SDK metrics port | `9091` | No | | `POD_ID` | Kubernetes pod identifier | `None` | No | -| `SIENTIA_MINIO_RETENTION_HOURS` | Retention window for offloaded MinIO objects | `168` | No | -| `SIENTIA_MINIO_OFFLOAD_THRESHOLD_BYTES` | Offload threshold for DataFrame-derived payloads | `int(1.5 * 1024 * 1024)` | No | +| `MINIO_ENDPOINT_URL` | MinIO endpoint URL | `http://localhost:9000` | Yes | +| `MINIO_ACCESS_KEY` | MinIO access key | `minioadmin` | Yes | +| `MINIO_SECRET_KEY` | MinIO secret key | `minioadmin` | Yes | +| `MINIO_REGION_NAME` | MinIO region name | `us-east-1` | No | +| `MINIO_DEFAULT_BUCKET` | Default MinIO bucket | `laborious` | No | +| `MINIO_RETENTION_HOURS` | Retention window (hours) for offloaded MinIO objects | `24` | No | +| `SIENTIA_MINIO_OFFLOAD_THRESHOLD_MEGABYTES` | Offload threshold for DataFrame-derived payloads | `int(1.5 * 1024 * 1024)` | No | @@ -728,7 +841,7 @@ Laborious uses MinIO to prevent Temporal workflow history from carrying very lar Whenever a payload exceeds a configurable size threshold, it is stored as a parquet file in MinIO and the workflow history only keeps a lightweight reference. Notes: -- `SIENTIA_MINIO_OFFLOAD_THRESHOLD_BYTES` supports: +- `SIENTIA_MINIO_OFFLOAD_THRESHOLD_MEGABYTES` supports: - Integer bytes (e.g. `"1572864"`) - Float MiB (e.g. `"1.5"`), converted to bytes as `MiB * 1024 * 1024` - Fallback behavior uses `1.5 MiB` when the env var is missing or invalid. @@ -751,7 +864,7 @@ After evaluation, the payload is serialized for Temporal as a flat dict: - `data` is omitted / set to `None`. When an activity needs pandas operations, it resolves references using: -- `MinioDataFramePayload.dataframe_from_wire(...)` +- `MinioDataFramePayload.retrieve(minio_repo)` — downloads from MinIO or returns inline data as a DataFrame #### MinIO Object Naming (Retention Parsing) @@ -952,27 +1065,35 @@ This is the configuration created by the Orchestrator in Temporal. ``` laborious/ ├── activities/ # Temporal activity implementations -│ ├── activities.py # Main activities orchestrator +│ ├── activities.py # Main activities aggregator │ ├── gates.py # Data quality gates and filtering │ ├── mlflow.py # MLFlow model operations +│ ├── storage.py # PostgreSQL queries and MinIO offload +│ ├── model_metrics.py # Drift and regression metrics │ ├── opc.py # OPC server operations │ └── api.py # PI Web API operations ├── workflows/ # Temporal workflow definitions │ ├── predictions_batch.py # Main batch prediction workflow │ ├── minimal_retrain.py # Model retraining workflow +│ ├── drift.py # Data drift detection workflow +│ ├── simple_metrics.py # Regression metrics workflow │ └── sub_workflows/ # Sub-workflow implementations │ ├── prediction_process.py # Core prediction workflow │ └── format_and_export_prediction.py # Export workflow ├── worker/ # Worker implementation -│ └── worker.py # Main worker orchestrator +│ ├── worker.py # Main worker orchestrator +│ └── prepare_worker.py # Worker factory with autoscaling config ├── utils/ # Utility functions -│ ├── connectors_config.py # Database configuration +│ ├── connectors_config.py # Environment-driven config builders +│ ├── models/ # Data models +│ │ └── minio_dataframe_payload.py # MinIO-offloaded DataFrame payload │ ├── filters/ # Data quality filters │ │ ├── conditional_filters.py # Conditional data filters │ │ └── mlflow_filters.py # MLFlow response filters │ └── repository/ # Data access layer │ ├── model_repository.py # MLFlow model operations -│ └── opc_repository.py # OPC server operations +│ ├── opc_repository.py # OPC server operations +│ └── minio_manager.py # MinIO object storage operations ├── metrics.py # Prometheus metrics definitions └── __init__.py ``` diff --git a/laborious/utils/models/minio_dataframe_payload.py b/laborious/utils/models/minio_dataframe_payload.py index 4a51bb3..7fd9bc8 100644 --- a/laborious/utils/models/minio_dataframe_payload.py +++ b/laborious/utils/models/minio_dataframe_payload.py @@ -31,7 +31,7 @@ _OBJECT_TIMESTAMP_PATTERN = re.compile( ) OFFLOAD_THRESHOLD_BYTES = int( - float(getenv('SIENTIA_MINIO_OFFLOAD_THRESHOLD_BYTES', '1.5')) * 1024 * 1024 + float(getenv('SIENTIA_MINIO_OFFLOAD_THRESHOLD_MEGABYTES', '1.5')) * 1024 * 1024 ) # Relative prefix used for storing offloaded training datasets in MinIO. diff --git a/tests/conftest.py b/tests/conftest.py index 0cf3b48..7671dce 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,9 +2,9 @@ import os import sys from unittest.mock import MagicMock -# The production code converts SIENTIA_MINIO_OFFLOAD_THRESHOLD_BYTES to int at import-time. +# The production code converts SIENTIA_MINIO_OFFLOAD_THRESHOLD_MEGABYTES to int at import-time. # Tests must set it to a valid integer string to avoid import errors. -os.environ.setdefault('SIENTIA_MINIO_OFFLOAD_THRESHOLD_BYTES', '1') +os.environ.setdefault('SIENTIA_MINIO_OFFLOAD_THRESHOLD_MEGABYTES', '1') class DummyMinioDataFramePayload: diff --git a/values.yaml b/values.yaml index 08fa8d3..f1313d4 100644 --- a/values.yaml +++ b/values.yaml @@ -231,6 +231,10 @@ env: value: "sa-east-1" - name: MINIO_DEFAULT_BUCKET value: "sientia" + - name: MINIO_RETENTION_HOURS + value: "24" + - name: SIENTIA_MINIO_OFFLOAD_THRESHOLD_MEGABYTES + value: "1.5" - name: PI_WEB_API_BASE_URL value: "https://pivision.votorantimcimentos.com/piwebapi"