From c8b809f18994cccaa5a7f6c9461b639abd51eff2 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 17 Nov 2025 09:58:55 -0300 Subject: [PATCH] SIENTIAPDE-1273 Update dependencies and refactor data handling in various modules - Updated sientia-dataops-library dependency version from 1.5.3 to 1.5.4 in requirements files. - Updated sientia-mlops-library dependency version from 0.39.0 to 0.40.2 in requirements files. - Refactored return types in Gates, MLFlow, and ModelMetrics classes to return dictionaries instead of DataFrames for improved compatibility with downstream systems. - Removed the temporal_codec module as it is no longer needed for DataFrame serialization. - Adjusted data handling in the Drift workflow to ensure proper data structure is maintained. --- laborious/activities/gates.py | 16 +-- laborious/activities/mlflow.py | 6 +- laborious/activities/model_metrics.py | 16 +-- laborious/utils/temporal_codec.py | 149 -------------------------- laborious/worker/worker.py | 5 - laborious/workflows/drift.py | 4 +- requirements-light.txt | 2 +- requirements.txt | 4 +- tests.ipynb | 15 ++- 9 files changed, 40 insertions(+), 177 deletions(-) delete mode 100644 laborious/utils/temporal_codec.py diff --git a/laborious/activities/gates.py b/laborious/activities/gates.py index 7c8ae76..2ac33f1 100644 --- a/laborious/activities/gates.py +++ b/laborious/activities/gates.py @@ -405,7 +405,7 @@ class Gates(SientiaMonitoring): @activity.defn(name='format_transformed_data') - async def format_transformed_data(self, input_data: dict[str, Any]) -> DataFrame: + async def format_transformed_data(self, input_data: dict[str, Any]) -> dict[str, Any]: """ Format transformed data according to configured storage policies. """ @@ -423,10 +423,10 @@ class Gates(SientiaMonitoring): data = data.melt(id_vars='timestamp', var_name='variable', value_name='value') data['model_id'] = model_id - return data + return data.to_dict() @activity.defn(name='format_prediction') - async def format_prediction(self, input_data: dict[str, Any]) -> DataFrame: + async def format_prediction(self, input_data: dict[str, Any]) -> dict[str, Any]: """ Format prediction data according to configured storage policies. @@ -496,10 +496,10 @@ class Gates(SientiaMonitoring): self.info(f'Prediction formatted: {len(data)} rows', metadata) self.debug(f'Prediction data: {data.head(5).to_string()}', metadata) - return data + return data.to_dict() @activity.defn(name='format_default_prediction') - async def format_default_prediction(self, input_data: dict[str, Any]) -> DataFrame: + async def format_default_prediction(self, input_data: dict[str, Any]) -> dict[str, Any]: """ Create and format default prediction data for error conditions. @@ -541,10 +541,10 @@ class Gates(SientiaMonitoring): ) self.info(f'Default prediction formatted: {data.size} rows', metadata) - return data + return data.to_dict() @activity.defn(name='format_retrain_report') - async def format_retrain_report(self, input_data: dict[str, Any]) -> DataFrame: + async def format_retrain_report(self, input_data: dict[str, Any]) -> dict[str, Any]: """ Format retrain report data according to configured storage policies. """ @@ -573,7 +573,7 @@ class Gates(SientiaMonitoring): self.debug(f'Retrain report: {report.to_csv()}', metadata) - return report + return report.to_dict() @activity.defn(name='get_last_timestamp') async def get_last_timestamp(self, input_data: dict[str, Any]) -> str: diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index 0447215..129fb75 100644 --- a/laborious/activities/mlflow.py +++ b/laborious/activities/mlflow.py @@ -423,7 +423,7 @@ class MLFlow(SientiaMonitoring): @activity.defn(name='get_reference_data') - async def get_reference_data(self, input_data: dict[str, Any]) -> DataFrame | None: + async def get_reference_data(self, input_data: dict[str, Any]) -> list[dict[Hashable, Any]] | None: """ Get reference data from the MLflow Model Registry. @@ -433,7 +433,7 @@ class MLFlow(SientiaMonitoring): - model_name (str): Name of the MLFlow model to get reference data from Returns: - dict[Hashable, Any] | None: Reference data from the MLflow Model Registry. + list[dict[Hashable, Any]] | None: Reference data from the MLflow Model Registry. """ @@ -452,4 +452,4 @@ class MLFlow(SientiaMonitoring): reference_data['timestamp'] = to_datetime(reference_data['timestamp']) reference_data['timestamp'] = reference_data['timestamp'].dt.strftime(DATETIME_FORMAT) - return reference_data \ No newline at end of file + return reference_data.to_dict(orient='records') \ No newline at end of file diff --git a/laborious/activities/model_metrics.py b/laborious/activities/model_metrics.py index 9896fa7..78e7aad 100644 --- a/laborious/activities/model_metrics.py +++ b/laborious/activities/model_metrics.py @@ -138,7 +138,7 @@ class ModelMetrics(SientiaMonitoring): @activity.defn(name='calculate_drift') - async def calculate_drift(self, input_data: dict[str, Any]) -> DataFrame | dict: + async def calculate_drift(self, input_data: dict[str, Any]) -> list[dict[Hashable, Any]]: """ Calculate drift metrics for a model. @@ -217,11 +217,11 @@ class ModelMetrics(SientiaMonitoring): level=NotificationLevel.ERROR, attachment_content=traceback.format_exc(), ) - return {} + return [] if drift_df.empty: self.warning('No drift metrics found', metadata) - return {} + return [] # Drop unnecessary columns drift_df.drop(columns=['p_value', 'chunk_start_date', 'chunk_end_date'], inplace=True) @@ -238,7 +238,7 @@ class ModelMetrics(SientiaMonitoring): if drift_df.empty: self.warning('No drift metrics found after dropping rows where timestamp is not in target data', metadata) - return {} + return [] # Rename columns to match database columns drift_df.rename(columns={ @@ -261,10 +261,12 @@ class ModelMetrics(SientiaMonitoring): 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) - return drift_df + return drift_df.to_dict(orient='records') - async def calculate_simple_metrics(self, input_data: dict[str, Any]) -> DataFrame: + @activity.defn(name='calculate_simple_metrics') + async def calculate_simple_metrics(self, input_data: dict[str, Any]) -> list[dict[Hashable, Any]]: """ Calculate simple metrics for a model. Metrics available are: - rmse @@ -342,6 +344,6 @@ class ModelMetrics(SientiaMonitoring): self.debug(f'Simple metrics dataframe: Size {data.shape} \n{data.head(5).to_string()}', metadata) - return data + return data.to_dict(orient='records') diff --git a/laborious/utils/temporal_codec.py b/laborious/utils/temporal_codec.py deleted file mode 100644 index 2c6d70a..0000000 --- a/laborious/utils/temporal_codec.py +++ /dev/null @@ -1,149 +0,0 @@ -""" -Temporal Codec for DataFrame Serialization - -This module provides a custom Temporal DataConverter that automatically -serializes pandas DataFrames to Parquet format and deserializes them back. - -The codec only handles DataFrames, leaving all other types to the default -Temporal serialization mechanism. -""" - -import io -from collections.abc import Sequence -from typing import Any, List, Optional, Type - -from temporalio.api.common.v1 import Payload -from temporalio.converter import DataConverter, PayloadConverter -from pandas import DataFrame, read_parquet - - -class DataFramePayloadConverter(PayloadConverter): - """ - Custom PayloadConverter that serializes pandas DataFrames to Parquet format. - - Only DataFrames are handled by this converter. All other types are passed - to the default Temporal serialization mechanism. - """ - - def __init__(self, default_payload_converter: PayloadConverter): - """ - Initialize the DataFrame payload converter. - - Args: - default_payload_converter: The default Temporal payload converter - to use for non-DataFrame types - """ - self._default = default_payload_converter - - def to_payloads(self, values: Sequence[Any]) -> List[Payload]: - """ - Convert values to Temporal Payloads. - - If a value is a pandas DataFrame, it is serialized to Parquet format. - Otherwise, the default converter is used. - - Args: - values: The values to serialize - - Returns: - List[Payload]: The serialized payloads - """ - payloads = [] - for value in values: - # Check if value is a DataFrame - try: - if isinstance(value, DataFrame): - buffer = io.BytesIO() - value.to_parquet(buffer, engine='pyarrow', index=True) - payloads.append(Payload( - metadata={"encoding": b"parquet-dataframe"}, - data=buffer.getvalue() - )) - continue - except ImportError: - # pandas not available, fall through to default - pass - except Exception: - # Error serializing DataFrame, fall through to default - pass - - # Use default converter for all other types - default_payloads = self._default.to_payloads([value]) - payloads.extend(default_payloads) - - return payloads - - def from_payloads( - self, - payloads: Sequence[Payload], - type_hints: Optional[List[Type]] = None, - ) -> List[Any]: - """ - Convert Temporal Payloads back to Python values. - - If a payload metadata indicates it's a Parquet-serialized DataFrame, - it is deserialized. Otherwise, the default converter is used. - - Args: - payloads: The payloads to deserialize - type_hints: Optional type hints for the expected return types - - Returns: - List[Any]: The deserialized values - """ - values = [] - for i, payload in enumerate(payloads): - # Check if this is a Parquet-serialized DataFrame - if payload.metadata.get("encoding") == b"parquet-dataframe": - try: - buffer = io.BytesIO(payload.data) - values.append(read_parquet(buffer)) - continue - except ImportError: - # pandas not available, fall through to default - pass - except Exception: - # Error deserializing DataFrame, fall through to default - pass - - # Use default converter for all other types - # type_hints must have same length as payloads if provided - type_hint = type_hints[i] if type_hints and i < len(type_hints) else None - default_values = self._default.from_payloads([payload], [type_hint] if type_hint is not None else None) - values.extend(default_values) - - return values - - -def create_dataframe_data_converter() -> DataConverter: - """ - Create a DataConverter with DataFrame serialization support. - - Returns: - DataConverter: A DataConverter that handles DataFrames automatically - """ - # Get default converter to use as fallback - # DataConverter.default is an attribute, not a method - default_converter = DataConverter.default - - # Create a factory class that extends PayloadConverter - class DataFramePayloadConverterFactory(PayloadConverter): - def __init__(self): - super().__init__() - # Create instance of default payload converter to use as fallback - self._default_converter = default_converter.payload_converter_class() - - def to_payloads(self, values: Sequence[Any]) -> List[Payload]: - return DataFramePayloadConverter(self._default_converter).to_payloads(values) - - def from_payloads( - self, - payloads: Sequence[Payload], - type_hints: Optional[List[Type]] = None, - ) -> List[Any]: - return DataFramePayloadConverter(self._default_converter).from_payloads(payloads, type_hints) - - return DataConverter( - payload_converter_class=DataFramePayloadConverterFactory - ) - diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index 4700c48..1d28016 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -48,7 +48,6 @@ with workflow.unsafe.imports_passed_through(): build_opc_config, build_postgres_config, ) - from laborious.utils.temporal_codec import create_dataframe_data_converter from laborious.workflows.minimal_retrain import MinimalRetrain from laborious.workflows.predictions_batch import PredictionsBatch from laborious.workflows.drift import Drift @@ -133,14 +132,10 @@ async def main(): logger.custom_info(f'Starting Temporal Client at {host}...', metadata) - # Create custom data converter with DataFrame support - data_converter = create_dataframe_data_converter() - temporal_client = await client.Client.connect( target_host=host, namespace=os.getenv('TEMPORAL_NAMESPACE', 'laborious'), runtime=new_runtime, - data_converter=data_converter, ) logger.custom_info('Starting Workers...', metadata) diff --git a/laborious/workflows/drift.py b/laborious/workflows/drift.py index 043ef7e..60bb274 100644 --- a/laborious/workflows/drift.py +++ b/laborious/workflows/drift.py @@ -32,6 +32,8 @@ class Drift: } } + print(f'Input data: {input_data}', metadata) + model_config = input_data['model_config'] target_name = model_config['target'] @@ -93,7 +95,7 @@ class Drift: Activities.export_data_to_postgres, { **metadata, - 'data': drift_data.to_dict(orient='records'), + 'data': drift_data, 'schema': input_data['schema'], 'table_name': input_data['target_table_name'], 'timestamp_conversion': {'column': 'timestamp', 'format': DATETIME_FORMAT_WITH_TZ}, diff --git a/requirements-light.txt b/requirements-light.txt index e41ba53..26f2290 100644 --- a/requirements-light.txt +++ b/requirements-light.txt @@ -3,7 +3,7 @@ psycopg2-binary sqlalchemy asyncua redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.5.2 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.5.4 prometheus-client botocore boto3 diff --git a/requirements.txt b/requirements.txt index 750ef66..214e910 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,8 +3,8 @@ psycopg2-binary sqlalchemy asyncua redis -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.5.3 -git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.39.0 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.5.4 +git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.40.2 prometheus-client botocore boto3 diff --git a/tests.ipynb b/tests.ipynb index 25f1cf0..6933169 100644 --- a/tests.ipynb +++ b/tests.ipynb @@ -591,10 +591,19 @@ }, { "cell_type": "code", - "execution_count": 10, + "execution_count": 1, "id": "486b95b3", "metadata": {}, "outputs": [ + { + "data": { + "text/plain": [ + "[]" + ] + }, + "metadata": {}, + "output_type": "display_data" + }, { "data": { "text/html": [ @@ -724,6 +733,10 @@ "source": [ "from pandas import DataFrame\n", "\n", + "data = DataFrame()\n", + "\n", + "display(data.to_dict(orient='records'))\n", + "\n", "data = DataFrame({\n", " \"a\": {\"2025-01-01\": 1, \"2025-01-02\": 2, \"2025-01-03\": 3},\n", " \"b\": {\"2025-01-01\": 4, \"2025-01-02\": 5, \"2025-01-03\": 6},\n",