805 lines
33 KiB
Python
805 lines
33 KiB
Python
from sientia_do.repository.minio_repository import MinioRepository
|
|
from temporalio import activity, workflow
|
|
|
|
from laborious.utils.repository.minio_manager import MinioManager
|
|
|
|
with workflow.unsafe.imports_passed_through():
|
|
import traceback
|
|
from collections.abc import Callable, Mapping
|
|
from typing import Any
|
|
|
|
from pandas import DataFrame
|
|
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
|
|
from sientia_do.notifications.models import NotificationLevel
|
|
from sientia_do.observability.logger import Logger
|
|
from sientia_do.observability.metrics_controller import MetricsController
|
|
from sientia_do.utils.formatters import create_sample_dict
|
|
|
|
from laborious import metrics
|
|
from laborious.utils.dataframe_debug import build_dataframe_debug_message
|
|
from laborious.utils.filters.conditional_filters import (
|
|
filter_empty_data,
|
|
filter_specific_variables_null_values,
|
|
)
|
|
from laborious.utils.filters.mlflow_filters import api_error_filter, nan_values_filter
|
|
from laborious.utils.models.minio_dataframe_payload import MinioDataFramePayload
|
|
|
|
# Strongly-typed filter function signatures
|
|
InputFilterFunc = Callable[[DataFrame, dict[str, Any]], bool]
|
|
ResponseFilterFunc = Callable[[dict[str, Any], dict[str, Any]], bool]
|
|
ContentFilterFunc = Callable[[DataFrame, dict[str, Any]], bool]
|
|
|
|
# Input filter function mappings
|
|
input_filter_functions: dict[str, InputFilterFunc] = {
|
|
'SPECIFIC_VARIABLES_NULL_VALUES': filter_specific_variables_null_values,
|
|
'EMPTY_DATA': filter_empty_data,
|
|
}
|
|
|
|
# Confidence mappings kept separate from function maps to avoid Union types
|
|
input_path_confidence: Mapping[str, int] = {
|
|
'STOP': -1,
|
|
'CONTINUE': 2,
|
|
'REPEAT': -1,
|
|
}
|
|
|
|
# MLFlow response filter function mappings
|
|
mlflow_response_filter_functions: dict[str, ResponseFilterFunc] = {
|
|
'API_ERROR': api_error_filter,
|
|
}
|
|
|
|
mlflow_response_path_confidence: Mapping[str, int] = {
|
|
'STOP': -1,
|
|
'CONTINUE': 10,
|
|
'REPEAT': -1,
|
|
}
|
|
|
|
# MLFlow content filter function mappings
|
|
mlflow_content_filter_functions: dict[str, ContentFilterFunc] = {
|
|
'NAN_VALUES': nan_values_filter,
|
|
'EMPTY_DATA': filter_empty_data,
|
|
}
|
|
|
|
mlflow_content_path_confidence: Mapping[str, int] = {
|
|
'STOP': -1,
|
|
'CONTINUE': 18,
|
|
'REPEAT': -1,
|
|
}
|
|
|
|
|
|
class Gates(MinioManager):
|
|
"""
|
|
Data quality gates and filtering activities for the Laborious system.
|
|
|
|
This class implements comprehensive data quality validation and filtering
|
|
mechanisms that can be applied at different stages of the prediction pipeline.
|
|
It provides configurable filters with policy-based decision making to ensure
|
|
data integrity and quality throughout the ML workflow.
|
|
|
|
The class supports multiple filter types and implements a flexible policy
|
|
system that can be configured for different validation requirements. Each
|
|
filter returns a path decision (STOP, CONTINUE, REPEAT) along with confidence
|
|
scores and detailed comments for monitoring and debugging.
|
|
|
|
Attributes:
|
|
input_filter_functions (dict): Mapping of input filter names to functions
|
|
mlflow_response_filter_functions (dict): Mapping of MLFlow response filter names to functions
|
|
mlflow_content_filter_functions (dict): Mapping of MLFlow content filter names to functions
|
|
"""
|
|
|
|
minio_repository: MinioRepository | None = None
|
|
_MAX_DEBUG_DATAFRAME_ROWS = 100
|
|
|
|
def __init__(
|
|
self,
|
|
minio_repository: MinioRepository | None = None,
|
|
logger: Logger | None = None,
|
|
notification_handler: NotificationHandler | None = None,
|
|
metrics_controller: MetricsController | None = None,
|
|
):
|
|
"""
|
|
Initialize data quality gates with logging and notification capabilities.
|
|
|
|
Args:
|
|
logger: Logger instance for observability and debugging
|
|
notification_handler: Notification handler for alerts and monitoring
|
|
|
|
Raises:
|
|
Exception: If BaseActivity initialization fails
|
|
"""
|
|
MinioManager.__init__(
|
|
self, minio_repository, logger, notification_handler, metrics_controller
|
|
)
|
|
|
|
def close(self) -> None:
|
|
"""
|
|
Close the gates activity and clean up resources.
|
|
"""
|
|
|
|
MinioManager.close(self)
|
|
|
|
def __del__(self):
|
|
self.close()
|
|
|
|
def _debug_dataframe(self, message: str, data: Any, metadata: dict[str, Any]) -> None:
|
|
"""
|
|
Log dataframe content only when row count is below the configured threshold
|
|
|
|
Args:
|
|
- message (str): Base log message to identify the dataframe in logs
|
|
- data (Any): Dataframe-like payload to be logged
|
|
- metadata (dict[str, Any]): Workflow metadata for contextual logging
|
|
"""
|
|
self.debug(
|
|
build_dataframe_debug_message(
|
|
message=message,
|
|
data=data,
|
|
max_rows=self._MAX_DEBUG_DATAFRAME_ROWS,
|
|
),
|
|
metadata,
|
|
)
|
|
|
|
@staticmethod
|
|
def _read_filter_entry(config: dict[str, Any]) -> tuple[str, dict[str, Any]]:
|
|
"""
|
|
Read filter policy/config keys in a case-insensitive way.
|
|
|
|
Args:
|
|
config (dict[str, Any]): Filter configuration dictionary.
|
|
|
|
Return:
|
|
tuple[str, dict[str, Any]]: Parsed policy and config payload.
|
|
"""
|
|
normalized = {str(key).upper(): value for key, value in config.items()}
|
|
policy = normalized['POLICY']
|
|
filter_config = normalized.get('CONFIG', {})
|
|
return policy, filter_config
|
|
|
|
@activity.defn(name='input_gate')
|
|
async def input_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
|
|
"""
|
|
Apply input data quality filters and validation.
|
|
|
|
This activity validates input data quality using configurable filters
|
|
before proceeding with ML operations. It applies multiple filter types
|
|
and returns a path decision based on the filter results and configured
|
|
policies.
|
|
|
|
The method implements a comprehensive filtering system that:
|
|
1. Applies configured filters to input data
|
|
2. Evaluates filter results against policy configurations
|
|
3. Determines appropriate path decisions (STOP, CONTINUE, REPEAT)
|
|
4. Provides confidence scores and detailed comments
|
|
5. Handles errors gracefully with notification integration
|
|
|
|
Args:
|
|
input_data: Configuration and data for input validation
|
|
Required keys:
|
|
- metadata (dict): Workflow execution metadata
|
|
- filters (dict): Filter configuration and policies
|
|
- data (dict): Input data to validate
|
|
- path_priority (list[str]): Priority order for path decisions
|
|
|
|
Returns:
|
|
tuple: (path_flag, confidence, comment)
|
|
- path_flag (str | None): Decision path (STOP, CONTINUE, REPEAT, or None)
|
|
- confidence (int): Confidence score for the decision
|
|
- comment (str): Detailed explanation of the decision
|
|
|
|
Raises:
|
|
Exception: If filter execution fails or configuration is invalid
|
|
"""
|
|
metadata = input_data['metadata']
|
|
|
|
self.info('Performing input gate...', metadata)
|
|
|
|
filters = input_data['filters']
|
|
payload = MinioDataFramePayload.from_dict(input_data['data'])
|
|
data = await payload.retrieve(self.minio_repository, metadata)
|
|
path_priority = input_data['path_priority']
|
|
|
|
filter_output = []
|
|
|
|
self._debug_dataframe('Input data:', data, metadata)
|
|
self.debug(f'Filters: {filters}', metadata)
|
|
|
|
# Apply each configured filter
|
|
for fil, config in filters.items():
|
|
if fil not in input_filter_functions:
|
|
self.error(f'Filter {fil} not found', metadata)
|
|
continue
|
|
policy, filter_config = self._read_filter_entry(config)
|
|
try:
|
|
if input_filter_functions[fil](data, filter_config):
|
|
self.debug(f'Data not passed the input filter {fil}:{config}', metadata)
|
|
filter_output.append(policy)
|
|
except Exception as e:
|
|
trace = traceback.format_exc()
|
|
await self.send_notification_async(
|
|
metadata=metadata,
|
|
notification_id=f'INTPUT_GATE_ERROR__{fil}',
|
|
message=f'Error in filter {fil}:{config}: \n {e}',
|
|
block='input_gate',
|
|
level=NotificationLevel.ERROR,
|
|
attachment_content=trace,
|
|
)
|
|
|
|
for path_flag in path_priority:
|
|
if path_flag in filter_output:
|
|
self.info(f'Input gate result: {path_flag}', metadata)
|
|
return path_flag, input_path_confidence[path_flag], 'Input data with bad quality'
|
|
|
|
self.info('Nothing was filtered by the input gate', metadata)
|
|
|
|
del data
|
|
|
|
return None, 0, ''
|
|
|
|
@activity.defn(name='mlflow_response_gate')
|
|
async def mlflow_response_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
|
|
"""
|
|
Validate MLFlow API response quality and integrity.
|
|
|
|
This activity validates MLFlow API responses to ensure they meet quality
|
|
standards before proceeding with further processing. It applies response-specific
|
|
filters and determines appropriate path decisions based on response quality.
|
|
|
|
The method implements response validation that:
|
|
1. Applies MLFlow response-specific filters
|
|
2. Evaluates API response quality and integrity
|
|
3. Determines path decisions based on response validation results
|
|
4. Provides confidence scores and detailed validation comments
|
|
5. Handles API errors and response validation failures
|
|
|
|
Args:
|
|
input_data: Configuration and data for response validation
|
|
Required keys:
|
|
- metadata (dict): Workflow execution metadata
|
|
- filters (dict): Response filter configuration and policies
|
|
- data (dict): MLFlow API response data to validate
|
|
- type (str): Type of MLFlow operation (transform, predict)
|
|
- path_priority (list[str]): Priority order for path decisions
|
|
|
|
Returns:
|
|
tuple: (path_flag, confidence, comment)
|
|
- path_flag (str | None): Decision path (STOP, CONTINUE, REPEAT, or None)
|
|
- confidence (int): Confidence score for the decision
|
|
- comment (str): Detailed explanation of the decision
|
|
|
|
Raises:
|
|
Exception: If response validation fails or configuration is invalid
|
|
"""
|
|
metadata = input_data['metadata']
|
|
self.info('Performing mlflow response gate...', metadata)
|
|
raw_data = input_data['data']
|
|
filters = input_data['filters']
|
|
|
|
self.debug(
|
|
f'Input data: \n {create_sample_dict(raw_data, max_items=5, max_depth=5)}', metadata
|
|
)
|
|
self.debug(f'Filters: {filters}', metadata)
|
|
|
|
payload = MinioDataFramePayload.from_dict(raw_data)
|
|
data = await payload.retrieve(self.minio_repository, metadata)
|
|
|
|
gate_type = input_data['type']
|
|
path_priority = input_data['path_priority']
|
|
|
|
filter_output = []
|
|
|
|
comments = []
|
|
|
|
status = payload.status or {}
|
|
|
|
for fil, config in filters.items():
|
|
if fil not in mlflow_response_filter_functions:
|
|
continue
|
|
policy, filter_config = self._read_filter_entry(config)
|
|
try:
|
|
if mlflow_response_filter_functions[fil](status, filter_config):
|
|
filter_output.append(policy)
|
|
comments.append(status.get('message', 'Unknown MLFlow API error'))
|
|
await self.send_notification_async(
|
|
metadata=metadata,
|
|
notification_id=f'{gate_type.upper()}_GATE_RESPONSE_FILTER__{fil}',
|
|
message=status.get('message', 'Unknown MLFlow API error'),
|
|
block='mlflow_gate',
|
|
level=NotificationLevel.ERROR,
|
|
attachment_content=status.get('traceback'),
|
|
)
|
|
except Exception as e:
|
|
trace = traceback.format_exc()
|
|
await self.send_notification_async(
|
|
metadata=metadata,
|
|
notification_id=f'MLFLOW_GATE_RESPONSE_FILTER__{fil}',
|
|
message=f'Error in filter {fil}:{config}: \n {e}',
|
|
block='mlflow_gate',
|
|
level=NotificationLevel.ERROR,
|
|
attachment_content=trace,
|
|
)
|
|
|
|
for path_flag in path_priority:
|
|
if path_flag in filter_output:
|
|
self.info(f'Mlflow response gate result: {path_flag}', metadata)
|
|
return path_flag, mlflow_response_path_confidence[path_flag], ', '.join(comments)
|
|
|
|
self.info('Nothing was filtered by the mlflow response gate', metadata)
|
|
|
|
del data
|
|
|
|
return None, 0, ''
|
|
|
|
@activity.defn(name='mlflow_content_gate')
|
|
async def mlflow_content_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str]:
|
|
"""
|
|
Validate MLFlow prediction content quality and integrity.
|
|
|
|
This activity validates the content of MLFlow predictions to ensure they
|
|
meet quality standards before export and persistence. It applies content-specific
|
|
filters and determines appropriate path decisions based on content quality.
|
|
|
|
The method implements content validation that:
|
|
1. Applies MLFlow content-specific filters
|
|
2. Evaluates prediction content quality and integrity
|
|
3. Determines path decisions based on content validation results
|
|
4. Provides confidence scores and detailed validation comments
|
|
5. Handles content validation failures and quality issues
|
|
|
|
Args:
|
|
input_data: Configuration and data for content validation
|
|
Required keys:
|
|
- metadata (dict): Workflow execution metadata
|
|
- filters (dict): Content filter configuration and policies
|
|
- data (dict): MLFlow prediction content to validate
|
|
- type (str): Type of MLFlow operation (transform, predict)
|
|
- path_priority (list[str]): Priority order for path decisions
|
|
|
|
Returns:
|
|
tuple: (path_flag, confidence, comment)
|
|
- path_flag (str | None): Decision path (STOP, CONTINUE, REPEAT, or None)
|
|
- confidence (int): Confidence score for the decision
|
|
- comment (str): Detailed explanation of the decision
|
|
|
|
Raises:
|
|
Exception: If content validation fails or configuration is invalid
|
|
"""
|
|
metadata = input_data['metadata']
|
|
self.info('Performing mlflow content gate...', metadata)
|
|
|
|
filters = input_data['filters']
|
|
|
|
payload = MinioDataFramePayload.from_dict(input_data['data'])
|
|
data = await payload.retrieve(self.minio_repository, metadata)
|
|
|
|
gate_type = input_data['type']
|
|
path_priority = input_data['path_priority']
|
|
|
|
filter_output = []
|
|
|
|
self._debug_dataframe('Input data:', data, metadata)
|
|
self.debug(f'Filters: \n {filters}', metadata)
|
|
|
|
for fil, config in filters.items():
|
|
if fil not in mlflow_content_filter_functions:
|
|
continue
|
|
policy, filter_config = self._read_filter_entry(config)
|
|
try:
|
|
if mlflow_content_filter_functions[fil](data, filter_config):
|
|
filter_output.append(policy)
|
|
await self.send_notification_async(
|
|
metadata=metadata,
|
|
notification_id=f'{gate_type.upper()}_GATE_CONTENT_FILTER__{fil}',
|
|
message=f'Data not passed the content filter {fil}:{config}',
|
|
block='mlflow_gate',
|
|
level=NotificationLevel.WARNING,
|
|
attachment_content=data.to_string(),
|
|
)
|
|
except Exception as e:
|
|
trace = traceback.format_exc()
|
|
await self.send_notification_async(
|
|
metadata=metadata,
|
|
notification_id=f'MLFLOW_GATE_CONTENT_FILTER__{fil}',
|
|
message=f'Error in filter {fil}:{config}: \n {e}',
|
|
block='mlflow_gate',
|
|
level=NotificationLevel.ERROR,
|
|
attachment_content=trace,
|
|
)
|
|
|
|
for path_flag in path_priority:
|
|
if path_flag in filter_output:
|
|
self.info(f'Mlflow content gate result: {path_flag}', metadata)
|
|
return (
|
|
path_flag,
|
|
mlflow_content_path_confidence[path_flag],
|
|
'Transformed data not passed the content filter',
|
|
)
|
|
|
|
self.info('Nothing was filtered by the mlflow content gate', metadata)
|
|
|
|
del data
|
|
|
|
return None, 0, ''
|
|
|
|
def get_prediction_store_policy(
|
|
self, prediction_store_policy: str, metadata: dict[str, Any]
|
|
) -> tuple[str, int]:
|
|
"""
|
|
Parse and validate prediction store policy configuration.
|
|
|
|
This method parses prediction store policy strings in the format 'type:value'
|
|
and validates them against allowed policy types and values. It provides
|
|
sensible defaults for invalid configurations and logs policy validation
|
|
failures for operational monitoring.
|
|
|
|
Supported Policy Types:
|
|
- 'lts': Latest timestamp - sorts data by timestamp descending
|
|
- 'erl': Earliest timestamp - sorts data by timestamp ascending
|
|
|
|
Args:
|
|
prediction_store_policy (str): Policy string in format 'type:value'
|
|
metadata (dict[str, Any]): Context metadata for logging and notifications
|
|
|
|
Returns:
|
|
tuple[str, int]: (policy_type, policy_value)
|
|
- policy_type (str): Validated policy type ('lts' or 'erl')
|
|
- policy_value (int): Number of rows to retain
|
|
"""
|
|
policy_elements = prediction_store_policy.split(':')
|
|
|
|
if len(policy_elements) < 2:
|
|
self.error(
|
|
f'Invalid prediction store policy: {prediction_store_policy}, using default policy',
|
|
metadata,
|
|
)
|
|
return 'lts', 1
|
|
|
|
policy_type = policy_elements[0]
|
|
policy_value = policy_elements[1]
|
|
|
|
# If the policy_type is not lts or erl, we use the default policy
|
|
# If the policty_value is not a number or 0, we use the default policy
|
|
if (
|
|
policy_type not in ['lts', 'erl']
|
|
or not policy_value.isdigit()
|
|
or int(policy_value) == 0
|
|
):
|
|
self.error(
|
|
f'Invalid prediction store policy: {prediction_store_policy}, using default policy',
|
|
metadata,
|
|
)
|
|
return 'lts', 1
|
|
|
|
return policy_type, int(policy_value)
|
|
|
|
@activity.defn(name='format_transformed_data')
|
|
async def format_transformed_data(self, input_data: dict[str, Any]) -> MinioDataFramePayload:
|
|
"""
|
|
Format transformed data for storage and export operations.
|
|
|
|
This method formats transformed data from MLFlow model transformations
|
|
into a standardized format suitable for database storage. It converts
|
|
wide-format data (columns as variables) into long-format (melted)
|
|
with proper timestamp handling and model identification.
|
|
|
|
The formatting process includes:
|
|
1. Converting input data dictionary to DataFrame
|
|
2. Extracting timestamps from DataFrame index
|
|
3. Resetting index to create sequential row numbers
|
|
4. Melting data from wide format to long format (variable-value pairs)
|
|
5. Adding model_id for data lineage tracking
|
|
|
|
Args:
|
|
input_data (dict): Input data containing:
|
|
- metadata (dict): Workflow execution metadata
|
|
- data (dict[str, Any]): Transformed data to format (DataFrame-compatible dict)
|
|
- model_id (str): Unique identifier for the ML model
|
|
|
|
Returns:
|
|
dict: Formatted data dictionary with keys:
|
|
- timestamp (dict): Timestamp values indexed by row number
|
|
- variable (dict): Variable names indexed by row number
|
|
- value (dict): Variable values indexed by row number
|
|
- model_id (dict): Model identifiers indexed by row number
|
|
"""
|
|
metadata = input_data['metadata']
|
|
|
|
model_id = input_data['model_id']
|
|
|
|
self.info('Formatting transformed data...', metadata)
|
|
|
|
payload = MinioDataFramePayload.from_dict(input_data['data'])
|
|
data = await payload.retrieve(self.minio_repository, metadata)
|
|
|
|
data['timestamp'] = data.index
|
|
data = data.reset_index(drop=True)
|
|
|
|
data = data.melt(id_vars='timestamp', var_name='variable', value_name='value')
|
|
data['model_id'] = model_id
|
|
|
|
return await MinioDataFramePayload.from_dataframe(
|
|
dataframe=data,
|
|
minio_repo=self.minio_repository,
|
|
model_name=input_data['model_name'],
|
|
operation='transform',
|
|
workflow_metadata=metadata,
|
|
last_timestamp=payload.last_timestamp,
|
|
logger=self.logger,
|
|
)
|
|
|
|
@activity.defn(name='format_prediction')
|
|
async def format_prediction(self, input_data: dict[str, Any]) -> dict:
|
|
"""
|
|
Format prediction data according to configured storage policies.
|
|
|
|
This method formats prediction data for storage and export operations.
|
|
It applies timestamp-based sorting policies, adds metadata fields,
|
|
and ensures data consistency before persistence. The method supports
|
|
multiple storage policies for flexible data retention strategies.
|
|
|
|
If only one row is present, we use the last timestamp as the timestamp
|
|
|
|
Storage Policies:
|
|
- 'lts:N': Latest timestamp - retains N most recent predictions
|
|
- 'erl:N': Earliest timestamp - retains N oldest predictions
|
|
|
|
Args:
|
|
input_data (dict): Input data containing:
|
|
- data (dict[str, Any]): Raw prediction data to format
|
|
- timestamp (str): Timestamp of the data
|
|
- model_id (str): Unique identifier for the ML model
|
|
- prediction_confidence (float): Confidence score for the prediction
|
|
- prediction_store_policy (str): Storage policy in format 'type:value'
|
|
|
|
Returns:
|
|
dict: Formatted prediction data ready for storage and export
|
|
"""
|
|
metadata = input_data['metadata']
|
|
last_timestamp = input_data['timestamp']
|
|
prediction_store_policy = input_data['prediction_store_policy']
|
|
self.info('Formatting prediction...', metadata)
|
|
|
|
payload = MinioDataFramePayload.from_dict(input_data['data'])
|
|
data = await payload.retrieve(self.minio_repository, metadata)
|
|
|
|
# Create timestamp column from index and reset index
|
|
data['timestamp'] = data.index
|
|
data = data.reset_index(drop=True)
|
|
|
|
self.debug(f'Prediction store policy: {prediction_store_policy}', metadata)
|
|
self._debug_dataframe('Prediction data:', data, metadata)
|
|
|
|
policy_type, policy_value = self.get_prediction_store_policy(
|
|
prediction_store_policy, metadata
|
|
)
|
|
|
|
# If data has no timestamp, we use the default timestamp and not sort the data
|
|
self.info(
|
|
f'Sorting data by timestamp and applying policy: {policy_type}:{policy_value}', metadata
|
|
)
|
|
|
|
# If policy_type is lts, we need to sort the data by timestamp descending and take the first policy_value rows
|
|
if policy_type == 'lts':
|
|
self.debug('Sorting data by timestamp descending', metadata)
|
|
data = data.sort_values(by='timestamp', ascending=False)
|
|
# If policy_type is erl, we need to sort the data by timestamp ascending and take the first policy_value rows
|
|
elif policy_type == 'erl':
|
|
self.debug('Sorting data by timestamp ascending', metadata)
|
|
data = data.sort_values(by='timestamp', ascending=True)
|
|
else:
|
|
self.error(f'Invalid policy type: {policy_type}, using default policy', metadata)
|
|
raise ValueError(f'Invalid policy type: {policy_type}')
|
|
|
|
int_policy_value = int(policy_value)
|
|
|
|
data = data.head(int_policy_value)
|
|
if int_policy_value == 1:
|
|
data['timestamp'] = last_timestamp
|
|
|
|
data['model_id'] = input_data['model_id']
|
|
data['prediction_confidence'] = input_data['prediction_confidence']
|
|
data['prediction_status'] = 'Good'
|
|
data['comments'] = ''
|
|
data = data.sort_values(by='timestamp', ascending=False)
|
|
data = data.reset_index(drop=True)
|
|
|
|
self.info(f'Prediction formatted: {len(data)} rows', metadata)
|
|
self._debug_dataframe('Prediction data:', data, metadata)
|
|
|
|
return data.to_dict()
|
|
|
|
@activity.defn(name='format_default_prediction')
|
|
async def format_default_prediction(self, input_data: dict[str, Any]) -> dict:
|
|
"""
|
|
Create and format default prediction data for error conditions.
|
|
|
|
This method generates default prediction data when the main prediction
|
|
pipeline encounters errors or quality issues. It creates a standardized
|
|
data structure with zero values for predictions and useful metadata
|
|
for operational monitoring and debugging.
|
|
|
|
The default prediction serves as a fallback mechanism to:
|
|
1. Maintain data pipeline continuity during failures
|
|
2. Provide operational visibility into prediction quality issues
|
|
3. Enable downstream systems to handle error conditions gracefully
|
|
4. Support debugging and troubleshooting efforts
|
|
|
|
Args:
|
|
input_data (dict): Input data containing:
|
|
- timestamp (str): Timestamp for the default prediction
|
|
- model_id (str): Unique identifier for the ML model
|
|
- prediction_confidence (float): Confidence score (typically low for errors)
|
|
- comment (str): Error description or operational comment
|
|
|
|
Returns:
|
|
dict: Formatted default prediction data with error indicators
|
|
"""
|
|
|
|
metadata = input_data['metadata']
|
|
self.debug('Formatting default prediction...', metadata)
|
|
|
|
data = DataFrame(
|
|
{
|
|
'prediction': [0],
|
|
'response_time': [0],
|
|
'timestamp': [input_data['timestamp']],
|
|
'model_id': [input_data['model_id']],
|
|
'prediction_confidence': [input_data['prediction_confidence']],
|
|
'prediction_status': ['Bad'],
|
|
'comments': [input_data['comment']],
|
|
}
|
|
)
|
|
|
|
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:
|
|
"""
|
|
Format retrain report data for storage and audit trail maintenance.
|
|
|
|
This method formats model retraining operation results into a standardized
|
|
report format suitable for database storage and operational monitoring.
|
|
It captures retraining status, timestamps, and model version information
|
|
for comprehensive audit trails and operational visibility.
|
|
|
|
The formatting process includes:
|
|
1. Extracting retraining experiment response data
|
|
2. Capturing model update report information (version, MLflow IDs)
|
|
3. Formatting timestamps and status information
|
|
4. Conditionally including version information for successful retrains
|
|
|
|
Args:
|
|
input_data (dict): Input data containing:
|
|
- metadata (dict): Workflow execution metadata
|
|
- experiment_response (dict): Retraining experiment response containing:
|
|
- success (bool): Retraining operation success status
|
|
- timestamp (str): Timestamp of the retraining operation
|
|
- message (str): Status message or error description
|
|
- update_report (dict): Model update report containing:
|
|
- version (str): New model version identifier
|
|
- mlflow_run_id (str): MLflow run identifier
|
|
- mlflow_experiment_id (str): MLflow experiment identifier
|
|
- model_id (str): Unique identifier for the ML model
|
|
- model_name (str): Name of the ML model
|
|
|
|
Returns:
|
|
dict: Formatted retrain report dictionary with keys:
|
|
- model_id (dict): Model identifiers indexed by row number
|
|
- model_name (dict): Model names indexed by row number
|
|
- timestamp (dict): Retraining timestamps indexed by row number
|
|
- status (dict): Retraining status messages indexed by row number
|
|
- version (dict, optional): Model versions indexed by row number
|
|
Only included if experiment_response['success'] is True
|
|
- mlflow_run_id (dict, optional): MLflow run IDs indexed by row number
|
|
Only included if experiment_response['success'] is True
|
|
- mlflow_experiment_id (dict, optional): MLflow experiment IDs indexed by row number
|
|
Only included if experiment_response['success'] is True
|
|
"""
|
|
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_dataframe('Retrain report:', report, metadata)
|
|
|
|
return report.to_dict()
|
|
|
|
@activity.defn(name='write_metrics')
|
|
async def write_metrics(self, input_data: dict[str, Any]):
|
|
"""
|
|
Write prediction performance metrics to Prometheus monitoring system.
|
|
|
|
This method records comprehensive metrics for prediction operations,
|
|
enabling operational monitoring, performance analysis, and alerting.
|
|
It tracks prediction counts, confidence levels, and response times
|
|
for each model and pipeline combination.
|
|
|
|
Metrics Recorded:
|
|
1. Prediction Count: Incremental counter for successful predictions
|
|
2. Confidence Monitor: Current confidence level for predictions
|
|
3. Response Time Monitor: Histogram of prediction response times
|
|
|
|
Args:
|
|
input_data (dict): Input data containing:
|
|
- metadata (dict[str, Any]): Workflow execution metadata
|
|
- prediction (dict[str, Any]): Prediction data with metrics
|
|
|
|
Raises:
|
|
Exception: If metrics writing fails or configuration is invalid
|
|
"""
|
|
metadata = input_data['metadata']
|
|
prediction = DataFrame(input_data['prediction'])
|
|
prediction_confidence = prediction['prediction_confidence'].values[0]
|
|
response_time = prediction['response_time'].values[0]
|
|
opc_metrics = input_data['opc_metrics']
|
|
|
|
self.info(f'Writing metrics for model {metadata["model_name"]}', metadata)
|
|
|
|
core_tags = {
|
|
'pod_id': self.pod_id,
|
|
'runtime': self.runtime,
|
|
'operation_type': 'predict',
|
|
'model_name': metadata['model_name'],
|
|
'workflow_name': metadata['workflow_name'],
|
|
}
|
|
await self.emit_metric(
|
|
metric_object=metrics.PREDICTIONS_WRITTEN_COUNT,
|
|
tags=core_tags,
|
|
)
|
|
|
|
await self.emit_metric(
|
|
metric_object=metrics.PREDICTION_CONFIDENCE_MONITOR,
|
|
method='set',
|
|
tags=core_tags,
|
|
value=prediction_confidence,
|
|
)
|
|
|
|
await self.emit_metric(
|
|
metric_object=metrics.PREDICTION_RESPONSE_TIME_MONITOR,
|
|
method='observe',
|
|
tags=core_tags,
|
|
value=response_time,
|
|
)
|
|
|
|
for server_id, tags in opc_metrics.items():
|
|
for tag, response_time in tags.items():
|
|
if response_time is not None:
|
|
await self.emit_metric(
|
|
metric_object=metrics.PREDICTION_OPC_WRITING_RESPONSE_TIME_MONITOR,
|
|
method='observe',
|
|
tags={
|
|
**core_tags,
|
|
'opc_server_id': server_id,
|
|
'tag': tag,
|
|
},
|
|
value=response_time,
|
|
)
|
|
|
|
await self.emit_metric(
|
|
metric_object=metrics.PREDICTION_OPC_WRITING_COUNT,
|
|
tags={
|
|
**core_tags,
|
|
'opc_server_id': server_id,
|
|
'tag': tag,
|
|
},
|
|
)
|
|
|
|
self.info(f'Metrics written for model {metadata["model_name"]}', metadata)
|