From f84d38a837962c9cd9d5d5c63355e86d49448166 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 30 Mar 2026 08:51:41 -0300 Subject: [PATCH] SIENTIAPDE-1712 SIENTIAPDE-1712 Implement debug logging for DataFrames in MLFlow and MLFlowRepository classes. Added a method to log DataFrame content conditionally based on row count, enhancing debugging capabilities while preventing excessive log output. --- laborious/activities/mlflow.py | 34 ++++++++++++-- .../utils/repository/model_repository.py | 46 ++++++++++++++----- 2 files changed, 64 insertions(+), 16 deletions(-) diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index ad9b6ae..d1cb50a 100644 --- a/laborious/activities/mlflow.py +++ b/laborious/activities/mlflow.py @@ -42,6 +42,7 @@ class MLFlow(MinioManager): mlflow_password (str): MLFlow authentication password model_monitoring_repository (MLFlowRepository): Repository for MLFlow operations """ + _MAX_DEBUG_DATAFRAME_ROWS = 100 def __init__( self, @@ -94,6 +95,32 @@ class MLFlow(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 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}' + ), + metadata, + ) + @activity.defn(name='request_transform') async def request_transform(self, input_data: dict[str, Any]) -> MinioDataFramePayload: """ @@ -133,8 +160,7 @@ class MLFlow(MinioManager): model_name = input_data['model_name'] model_config = input_data.get('model_config', {}) - self.debug('Raw input data:', metadata) - self.debug(data.head(5).to_string(), metadata) + self._debug_dataframe('Raw input data:', data, metadata) # Sort by created_at in descending order and keep first occurrence of each variable/timestamp pair data = data.sort_values('created_at', ascending=False).drop_duplicates( @@ -150,7 +176,7 @@ class MLFlow(MinioManager): data['timestamp'] = data.index - self.debug(f'Processed input data: \n {data.to_csv()}', metadata) + self._debug_dataframe('Processed input data:', data, metadata) # Request transformation from MLFlow model response_data = await self.model_monitoring_repository.transform( @@ -231,7 +257,7 @@ class MLFlow(MinioManager): model_name = input_data['model_name'] model_config = input_data.get('model_config', {}) - self.debug(f'Input data for: \n {data.head(5).to_string()}', metadata) + self._debug_dataframe('Input data for prediction:', data, metadata) # Convert numpy.nan to None for model compatibility data.replace(np.nan, None, inplace=True) diff --git a/laborious/utils/repository/model_repository.py b/laborious/utils/repository/model_repository.py index fc965dd..d4b2ba5 100644 --- a/laborious/utils/repository/model_repository.py +++ b/laborious/utils/repository/model_repository.py @@ -64,6 +64,8 @@ def force_memory_release(logger: Logger): class MLFlowRepository(SientiaMonitoring): + _MAX_DEBUG_DATAFRAME_ROWS = 100 + def __init__( self, host: str, @@ -94,6 +96,32 @@ class MLFlowRepository(SientiaMonitoring): self._cache_lock = threading.RLock() self.logger = logger + 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 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}' + ), + metadata, + ) + """ Functions related to get model registry parameters """ @@ -1168,7 +1196,7 @@ class MLFlowRepository(SientiaMonitoring): and returned in the response structure rather than propagated. """ - self.debug(f'Data received for model transformation: {data.to_csv()}', metadata) + self._debug_dataframe('Data received for model transformation:', data, metadata) # data.to_csv( # f"tmp/data_{model_name}.csv", index=True) @@ -1186,9 +1214,8 @@ class MLFlowRepository(SientiaMonitoring): metadata=metadata, ) - self.debug( - f'Data received from model transformation: {transformed_data.head(5).to_csv()}', - metadata, + self._debug_dataframe( + 'Data received from model transformation:', transformed_data, metadata ) # transformed_data.to_csv( @@ -1254,9 +1281,7 @@ class MLFlowRepository(SientiaMonitoring): input_index = data.index start_time = datetime.now() - self.debug( - f'Data received for model prediction: {data.to_dict(orient="records")}', metadata - ) + self._debug_dataframe('Data received for model prediction:', data, metadata) # data.to_csv( # f"tmp/treated_data_{model_name}.csv", index=True) @@ -1273,10 +1298,7 @@ class MLFlowRepository(SientiaMonitoring): end_time = datetime.now() if isinstance(predict_data, pd.DataFrame): - self.debug( - f'Data received from model prediction: {predict_data.to_dict(orient="records")}', - metadata, - ) + self._debug_dataframe('Data received from model prediction:', predict_data, metadata) # predict_data.to_csv( # f"tmp/predicted_data_{model_name}.csv", index=True) @@ -1348,7 +1370,7 @@ class MLFlowRepository(SientiaMonitoring): """ self.info(f'Starting model retraining workflow for {model_name}', metadata) - self.debug(f'Data received for model retraining: {data.to_csv()}', metadata) + self._debug_dataframe('Data received for model retraining:', data, metadata) target_name = model_config.get('target', None)