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

@@ -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)

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)