SIENTIAPDE-1084
Update README.md to enhance installation instructions and refactor MLFlow filters - Added steps for installing GitHub CLI and authenticating with GitHub. - Updated the `api_error_filter` and `nan_values_filter` functions to improve parameter handling and streamline logic.
This commit is contained in:
vitor-aignosi
20
README.md
20
README.md
@@ -369,9 +369,23 @@ flowchart LR
|
||||
```
|
||||
|
||||
3. **Install dependencies**
|
||||
```bash
|
||||
pip install -r requirements.txt
|
||||
```
|
||||
|
||||
1. **Install github cli**
|
||||
```bash
|
||||
sudo apt update
|
||||
sudo apt install gh -y
|
||||
```
|
||||
|
||||
2. **Authenticate with github**
|
||||
```bash
|
||||
gh auth login
|
||||
```
|
||||
|
||||
3. **Run the install_dependencies.sh script**
|
||||
```bash
|
||||
chmod +x install_dependencies.sh
|
||||
./install_dependencies.sh
|
||||
```
|
||||
|
||||
4. **Create environment configuration file**
|
||||
```bash
|
||||
|
||||
@@ -1,240 +1,61 @@
|
||||
"""
|
||||
MLFlow Response Filters Module
|
||||
|
||||
This module provides MLFlow-specific data filtering functions for the Sientia DataOps Laborious system.
|
||||
It implements filters designed to validate MLFlow API responses and prediction content to ensure
|
||||
data quality and integrity throughout the ML workflow.
|
||||
|
||||
The module implements filters for:
|
||||
1. MLFlow API error detection and validation
|
||||
2. NaN value identification in prediction results
|
||||
3. Response content quality assessment
|
||||
4. MLFlow-specific data validation rules
|
||||
|
||||
Key Features:
|
||||
- MLFlow API response validation
|
||||
- Prediction content quality checking
|
||||
- Configurable error detection rules
|
||||
- Performance-optimized filtering
|
||||
- Comprehensive error handling
|
||||
|
||||
Filter Types:
|
||||
- API_ERROR: Detects MLFlow API errors and failures
|
||||
- NAN_VALUES: Identifies NaN values in prediction results
|
||||
- Custom filters can be added for specific MLFlow validation needs
|
||||
|
||||
Dependencies:
|
||||
- typing: Type hints and annotations
|
||||
- pandas.DataFrame: Data manipulation and processing (for some filters)
|
||||
"""
|
||||
|
||||
from typing import Any, Dict, Union
|
||||
import numpy as np
|
||||
from pandas import DataFrame
|
||||
|
||||
|
||||
def api_error_filter(data: Union[Dict[str, Any], Any], config: Dict[str, Any]) -> bool:
|
||||
def api_error_filter(response: dict, _config: dict) -> bool:
|
||||
"""
|
||||
Filter MLFlow API responses for error conditions.
|
||||
|
||||
|
||||
This function analyzes MLFlow API responses to detect error conditions
|
||||
and determine if the response should be filtered out due to quality
|
||||
or reliability issues.
|
||||
|
||||
The filter implements comprehensive error detection:
|
||||
1. HTTP error status code checking
|
||||
2. MLFlow error message detection
|
||||
3. Response structure validation
|
||||
4. Configurable error thresholds
|
||||
|
||||
|
||||
|
||||
Args:
|
||||
data: MLFlow API response data (dict or other types)
|
||||
config: Filter configuration dictionary
|
||||
response: MLFlow API response data (dict)
|
||||
_config: Filter configuration dictionary
|
||||
Required keys:
|
||||
- error_codes (list, optional): List of error codes to detect
|
||||
- error_keywords (list, optional): List of error keywords to detect
|
||||
- check_structure (bool, optional): Whether to validate response structure
|
||||
|
||||
|
||||
Returns:
|
||||
bool: True if data should be filtered (contains errors), False otherwise
|
||||
|
||||
Filter Logic:
|
||||
- Returns True (filter) if API errors are detected
|
||||
- Returns False (pass) if response is error-free
|
||||
- Handles various response formats gracefully
|
||||
- Supports configurable error detection rules
|
||||
|
||||
Example:
|
||||
>>> # Successful response
|
||||
>>> response = {'status': 'success', 'data': [1, 2, 3]}
|
||||
>>> config = {'error_keywords': ['error', 'failed', 'exception']}
|
||||
>>> result = api_error_filter(response, config)
|
||||
>>> print(result)
|
||||
False # Response passes filter
|
||||
|
||||
>>> # Error response
|
||||
>>> error_response = {'status': 'error', 'message': 'Model not found'}
|
||||
>>> result = api_error_filter(error_response, config)
|
||||
>>> print(result)
|
||||
True # Response fails filter (contains error)
|
||||
|
||||
>>> # Exception response
|
||||
>>> exception_response = {'exception': 'Connection timeout'}
|
||||
>>> result = api_error_filter(exception_response, config)
|
||||
>>> print(result)
|
||||
True # Response fails filter (contains exception)
|
||||
|
||||
Default Configuration:
|
||||
- error_codes: ['error', 'failed', 'exception', 'timeout']
|
||||
- error_keywords: ['error', 'failed', 'exception', 'timeout', 'not_found']
|
||||
- check_structure: True
|
||||
|
||||
Note:
|
||||
The filter is designed to be flexible and can handle various
|
||||
MLFlow response formats and error conditions.
|
||||
|
||||
"""
|
||||
# Get configuration with defaults
|
||||
error_codes = config.get('error_codes', ['error', 'failed', 'exception', 'timeout'])
|
||||
error_keywords = config.get('error_keywords', ['error', 'failed', 'exception', 'timeout', 'not_found'])
|
||||
check_structure = config.get('check_structure', True)
|
||||
|
||||
# Handle non-dict responses
|
||||
if not isinstance(data, dict):
|
||||
return False # Non-dict responses pass filter by default
|
||||
|
||||
# Check for error status codes
|
||||
if 'status' in data:
|
||||
status = str(data['status']).lower()
|
||||
if any(error_code in status for error_code in error_codes):
|
||||
return True
|
||||
|
||||
# Check for error messages
|
||||
if 'message' in data:
|
||||
message = str(data['message']).lower()
|
||||
if any(keyword in message for keyword in error_keywords):
|
||||
return True
|
||||
|
||||
# Check for exception fields
|
||||
if 'exception' in data:
|
||||
if not response:
|
||||
return True
|
||||
|
||||
# Check for error fields
|
||||
if 'error' in data:
|
||||
|
||||
if not response['success']:
|
||||
return True
|
||||
|
||||
# Check response structure if enabled
|
||||
if check_structure:
|
||||
# Look for common error indicators in response structure
|
||||
for key, value in data.items():
|
||||
if isinstance(value, str):
|
||||
value_lower = value.lower()
|
||||
if any(keyword in value_lower for keyword in error_keywords):
|
||||
return True
|
||||
|
||||
# Response passes error filter
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def nan_values_filter(data: Union[Dict[str, Any], Any], config: Dict[str, Any]) -> bool:
|
||||
def nan_values_filter(predictions: DataFrame, _config: dict) -> bool:
|
||||
"""
|
||||
Filter data for NaN (Not a Number) values.
|
||||
|
||||
|
||||
This function detects NaN values in MLFlow prediction results and
|
||||
determines if the data quality is sufficient for further processing
|
||||
or export operations.
|
||||
|
||||
The filter implements NaN detection for:
|
||||
1. Numeric data validation
|
||||
2. Prediction result quality checking
|
||||
3. Configurable NaN thresholds
|
||||
4. Multiple data type handling
|
||||
|
||||
|
||||
Args:
|
||||
data: Data to check for NaN values (dict, list, or other types)
|
||||
config: Filter configuration dictionary
|
||||
predictions: DataFrame containing prediction data to check for NaN values
|
||||
_config: Filter configuration dictionary
|
||||
Required keys:
|
||||
- max_nan_ratio (float, optional): Maximum allowed NaN value ratio (0.0 to 1.0)
|
||||
- max_nan_count (int, optional): Maximum allowed NaN value count
|
||||
- check_nested (bool, optional): Whether to check nested data structures
|
||||
|
||||
|
||||
Returns:
|
||||
bool: True if data should be filtered (too many NaN values), False otherwise
|
||||
|
||||
Filter Logic:
|
||||
- Returns True (filter) if NaN value thresholds are exceeded
|
||||
- Returns False (pass) if NaN values are within acceptable limits
|
||||
- Handles various data structures gracefully
|
||||
- Supports both ratio and count-based thresholds
|
||||
|
||||
Example:
|
||||
>>> # Data with acceptable NaN values
|
||||
>>> data = {'predictions': [1.0, 2.0, float('nan'), 4.0]}
|
||||
>>> config = {'max_nan_ratio': 0.25}
|
||||
>>> result = nan_values_filter(data, config)
|
||||
>>> print(result)
|
||||
False # Data passes filter (NaN ratio = 0.25, equals max)
|
||||
|
||||
>>> # Data with too many NaN values
|
||||
>>> data = {'predictions': [1.0, float('nan'), float('nan'), 4.0]}
|
||||
>>> config = {'max_nan_ratio': 0.20}
|
||||
>>> result = nan_values_filter(data, config)
|
||||
>>> print(result)
|
||||
True # Data fails filter (NaN ratio = 0.5, exceeds max of 0.2)
|
||||
|
||||
Default Configuration:
|
||||
- max_nan_ratio: 0.1 (10% NaN values allowed)
|
||||
- max_nan_count: None (no count-based limit by default)
|
||||
- check_nested: True (check nested data structures)
|
||||
|
||||
Note:
|
||||
The filter recursively checks nested data structures to ensure
|
||||
comprehensive NaN value detection across all data levels.
|
||||
|
||||
"""
|
||||
# Get configuration with defaults
|
||||
max_nan_ratio = config.get('max_nan_ratio', 0.1)
|
||||
max_nan_count = config.get('max_nan_count', None)
|
||||
check_nested = config.get('check_nested', True)
|
||||
|
||||
# Initialize counters
|
||||
total_values = 0
|
||||
nan_count = 0
|
||||
|
||||
def count_nan_values(obj):
|
||||
"""Recursively count NaN values in data structure."""
|
||||
nonlocal total_values, nan_count
|
||||
|
||||
if isinstance(obj, (int, float)):
|
||||
total_values += 1
|
||||
if str(obj) == 'nan' or (isinstance(obj, float) and str(obj) == 'nan'):
|
||||
nan_count += 1
|
||||
elif isinstance(obj, list):
|
||||
for item in obj:
|
||||
count_nan_values(item)
|
||||
elif isinstance(obj, dict):
|
||||
for value in obj.values():
|
||||
count_nan_values(value)
|
||||
elif check_nested and hasattr(obj, '__iter__') and not isinstance(obj, str):
|
||||
try:
|
||||
for item in obj:
|
||||
count_nan_values(item)
|
||||
except (TypeError, AttributeError):
|
||||
pass
|
||||
|
||||
# Count NaN values in data
|
||||
count_nan_values(data)
|
||||
|
||||
# Check if we have any values to analyze
|
||||
if total_values == 0:
|
||||
return False # No values to check, pass filter
|
||||
|
||||
# Calculate NaN ratio
|
||||
nan_ratio = nan_count / total_values
|
||||
|
||||
# Check ratio threshold
|
||||
if nan_ratio > max_nan_ratio:
|
||||
data = predictions.replace({None: np.nan}).drop(
|
||||
columns=['timestamp'], errors='ignore').infer_objects()
|
||||
|
||||
if data.isna().all().all():
|
||||
return True
|
||||
|
||||
# Check count threshold (if specified)
|
||||
if max_nan_count is not None and nan_count > max_nan_count:
|
||||
return True
|
||||
|
||||
# Data passes NaN filter
|
||||
|
||||
return False
|
||||
|
||||
8
requirements_prepared.txt
Normal file
8
requirements_prepared.txt
Normal file
@@ -0,0 +1,8 @@
|
||||
temporalio
|
||||
psycopg2-binary
|
||||
sqlalchemy
|
||||
asyncua
|
||||
redis
|
||||
git+https://github.com/Aignosi/sientia-dataops-library.git@1.4.4
|
||||
git+https://github.com/Aignosi/sientia-mlops-library.git@0.38.12
|
||||
prometheus-client
|
||||
Reference in New Issue
Block a user