diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index 12fafdf..34376b1 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -16,6 +16,7 @@ with workflow.unsafe.imports_passed_through(): 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, @@ -86,6 +87,7 @@ class Gates(MinioManager): """ minio_repository: MinioRepository | None = None + _MAX_DEBUG_DATAFRAME_ROWS = 100 def __init__( self, @@ -118,6 +120,24 @@ class Gates(MinioManager): 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]]: """ @@ -179,7 +199,7 @@ class Gates(MinioManager): filter_output = [] - self.debug(f'Input data: {data.head(5).to_string()}', metadata) + self._debug_dataframe('Input data:', data, metadata) self.debug(f'Filters: {filters}', metadata) # Apply each configured filter @@ -355,7 +375,7 @@ class Gates(MinioManager): filter_output = [] - self.debug(f'Input data:\n {data.head(5).to_string()}', metadata) + self._debug_dataframe('Input data:', data, metadata) self.debug(f'Filters: \n {filters}', metadata) for fil, config in filters.items(): @@ -544,7 +564,7 @@ class Gates(MinioManager): data = data.reset_index(drop=True) self.debug(f'Prediction store policy: {prediction_store_policy}', metadata) - self.debug(f'Prediction data: {data.head(5).to_string()}', metadata) + self._debug_dataframe('Prediction data:', data, metadata) policy_type, policy_value = self.get_prediction_store_policy( prediction_store_policy, metadata @@ -581,7 +601,7 @@ class Gates(MinioManager): data = data.reset_index(drop=True) self.info(f'Prediction formatted: {len(data)} rows', metadata) - self.debug(f'Prediction data: {data.head(5).to_string()}', metadata) + self._debug_dataframe('Prediction data:', data, metadata) return data.to_dict() @@ -696,7 +716,7 @@ class Gates(MinioManager): 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) + self._debug_dataframe('Retrain report:', report, metadata) return report.to_dict() diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index d1cb50a..10d3d8d 100644 --- a/laborious/activities/mlflow.py +++ b/laborious/activities/mlflow.py @@ -20,6 +20,7 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.utils.formatters import create_sample_dict from laborious.utils.models.minio_dataframe_payload import MinioDataFramePayload + from laborious.utils.dataframe_debug import build_dataframe_debug_message from laborious.utils.repository.minio_manager import MinioManager from laborious.utils.repository.model_repository import MLFlowRepository @@ -104,19 +105,11 @@ class MLFlow(MinioManager): - data (Any): Dataframe-like object expected to expose shape and to_csv - metadata (dict[str, Any]): Workflow metadata for contextual logging """ - if not hasattr(data, 'shape') or not hasattr(data, 'to_csv'): - self.debug(f'{message}\n{data}', metadata) - return - - rows = data.shape[0] - if rows <= self._MAX_DEBUG_DATAFRAME_ROWS: - self.debug(f'{message}\n{data.to_csv()}', metadata) - return - self.debug( - ( - f'{message} skipped because dataframe has {rows} rows ' - f'(max: {self._MAX_DEBUG_DATAFRAME_ROWS}). Shape: {data.shape}' + build_dataframe_debug_message( + message=message, + data=data, + max_rows=self._MAX_DEBUG_DATAFRAME_ROWS, ), metadata, ) diff --git a/laborious/activities/model_metrics.py b/laborious/activities/model_metrics.py index 29b2848..b336fbd 100644 --- a/laborious/activities/model_metrics.py +++ b/laborious/activities/model_metrics.py @@ -17,6 +17,7 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.temporal.constants import DATETIME_FORMAT, DATETIME_FORMAT_WITH_TZ from laborious import metrics + from laborious.utils.dataframe_debug import build_dataframe_debug_message warnings.filterwarnings('ignore', category=RuntimeWarning, message='Degrees of freedom <= 0') warnings.filterwarnings( @@ -26,6 +27,8 @@ warnings.filterwarnings( class ModelMetrics(SientiaMonitoring): """ + _MAX_DEBUG_DATAFRAME_ROWS = 100 + Metrics activities for the Laborious system. This class provides activities for writing metrics to the Prometheus monitoring system. @@ -48,6 +51,24 @@ class ModelMetrics(SientiaMonitoring): 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, + ) + async def get_drift_metrics( self, reference_data: DataFrame, @@ -78,14 +99,9 @@ class ModelMetrics(SientiaMonitoring): model_analysis = ModelAnalysis(config=config) - self.debug( - f'Reference data: Size {reference_data.shape} \n{reference_data.head(5).to_string()}', - metadata, - ) + self._debug_dataframe(f'Reference data: Size {reference_data.shape}', reference_data, metadata) - self.debug( - f'Target data: Size {target_data.shape} \n{target_data.head(5).to_string()}', metadata - ) + self._debug_dataframe(f'Target data: Size {target_data.shape}', target_data, metadata) core_labels = self.get_core_labels(metadata, operation_type='detect_univariate_drift') start_time = time.time() @@ -142,9 +158,7 @@ class ModelMetrics(SientiaMonitoring): await self.observe_lag(start_time, metrics.MODEL_ANALYZE_LAG, core_labels) await self.emit_metric(metric_object=metrics.MODEL_ANALYZE_COUNT, tags=core_labels) - self.debug( - f'Drift dataframe: Size {drift_df.shape} \n{drift_df.head(5).to_string()}', metadata - ) + self._debug_dataframe(f'Drift dataframe: Size {drift_df.shape}', drift_df, metadata) return drift_df @@ -274,11 +288,7 @@ class ModelMetrics(SientiaMonitoring): drift_df['timestamp'] = drift_df['timestamp'].dt.tz_localize('UTC') drift_df['timestamp'] = drift_df['timestamp'].dt.strftime(DATETIME_FORMAT_WITH_TZ) - self.debug( - f'Drift dataframe: Size {drift_df.shape} \n{drift_df.head(5).to_string()}', metadata - ) - - self.debug(f'Drift dataframe: {drift_df.head(5).to_string()}', metadata) + self._debug_dataframe(f'Drift dataframe: Size {drift_df.shape}', drift_df, metadata) return drift_df.to_dict(orient='records') @@ -347,8 +357,6 @@ class ModelMetrics(SientiaMonitoring): data['data_size'] = data_size data['interval_minutes'] = interval_minutes - self.debug( - f'Simple metrics dataframe: Size {data.shape} \n{data.head(5).to_string()}', metadata - ) + self._debug_dataframe(f'Simple metrics dataframe: Size {data.shape}', data, metadata) return data.to_dict(orient='records') diff --git a/laborious/utils/dataframe_debug.py b/laborious/utils/dataframe_debug.py new file mode 100644 index 0000000..995b606 --- /dev/null +++ b/laborious/utils/dataframe_debug.py @@ -0,0 +1,34 @@ +from typing import Any + +from pandas import DataFrame + +DEFAULT_MAX_DEBUG_DATAFRAME_ROWS = 100 + + +def build_dataframe_debug_message( + message: str, + data: Any, + max_rows: int = DEFAULT_MAX_DEBUG_DATAFRAME_ROWS, +) -> str: + """ + Build a safe debug message for dataframe payloads + + Args: + - message (str): Base message to identify the logged payload + - data (Any): Payload to evaluate for dataframe-aware logging + - max_rows (int): Maximum dataframe row count allowed for full payload logging + + Return: + Formatted debug message with full dataframe content or compact summary + """ + if not isinstance(data, DataFrame): + return f'{message} {data}' + + rows = data.shape[0] + if rows <= max_rows: + return f'{message}\n{data.to_csv()}' + + return ( + f'{message} skipped because dataframe has {rows} rows ' + f'(max: {max_rows}). Shape: {data.shape}' + ) diff --git a/laborious/utils/repository/model_repository.py b/laborious/utils/repository/model_repository.py index d4b2ba5..fb404b7 100644 --- a/laborious/utils/repository/model_repository.py +++ b/laborious/utils/repository/model_repository.py @@ -36,6 +36,7 @@ from sientia_do.observability.sientia_monitoring import SientiaMonitoring from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ from laborious import metrics +from laborious.utils.dataframe_debug import build_dataframe_debug_message ARTIFACTS_PATH = './tmp/artifacts' TRANSFORMED_COMPRESSED_PATH = 'artifacts/training_transformer.pkl' @@ -105,19 +106,11 @@ class MLFlowRepository(SientiaMonitoring): - data (Any): Dataframe-like object expected to expose shape and to_csv - metadata (dict[str, Any]): Metadata for contextual logging """ - if not isinstance(data, pd.DataFrame): - self.debug(f'{message} {data}', metadata) - return - - rows = data.shape[0] - if rows <= self._MAX_DEBUG_DATAFRAME_ROWS: - self.debug(f'{message}\n{data.to_csv()}', metadata) - return - self.debug( - ( - f'{message} skipped because dataframe has {rows} rows ' - f'(max: {self._MAX_DEBUG_DATAFRAME_ROWS}). Shape: {data.shape}' + build_dataframe_debug_message( + message=message, + data=data, + max_rows=self._MAX_DEBUG_DATAFRAME_ROWS, ), metadata, )