gitea_admin/sientia-dataops-laborious_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SIENTIAPDE-1273

Browse Source 
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.
This commit is contained in:
vitor-aignosi
2025-11-17 09:58:55 -03:00
parent d2b365a34d
commit c8b809f189
9 changed files with 40 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
laborious/activities/gates.py
View File

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

6
laborious/activities/mlflow.py
View File

@@ -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
return reference_data.to_dict(orient='records')

16
laborious/activities/model_metrics.py
View File

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

149
laborious/utils/temporal_codec.py
View File

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

5
laborious/worker/worker.py
View File

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

4
laborious/workflows/drift.py
View File

@@ -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},

2
requirements-light.txt
View File

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

4
requirements.txt
View File

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

15
tests.ipynb
View File

@@ -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",