fix-s3-categorical-integer-columns (#312)

Author: JKL98ISR

.github/workflows/license-check.yml

@@ -50,7 +50,7 @@ jobs:
         with:
           requirements: "backend/requirements-all.txt"
           fail: "Copyleft,Other,Error"
-          exclude: '(psycopg2.*2\.9\.3|fqdn.*1\.5\.1|pyzmq.*25\.1\.2|debugpy.*1\.6\.7|certifi.*2023\.11\.17|tqdm.*4\.66\.1|webencodings.*0\.5\.1|torch.*1\.10\.2.*|torch.*1\.11\.0.*|pytorch-ignite.*0\.4\.10.*|torchaudio.*0\.11\.0.*|torchvision.*0\.12\.0.*|terminado.*0\.15\.0|qudida.*0\.0\.4|expiringdict.*1\.2\.2|botocore.*1\.29\.80|orderedmultidict.*1\.0\.1|deepchecks.*)'
+          exclude: '(envier.*0\.5\.0|psycopg2.*2\.9\.3|fqdn.*1\.5\.1|pyzmq.*25\.1\.2|debugpy.*1\.6\.7|certifi.*2023\.11\.17|tqdm.*4\.66\.1|webencodings.*0\.5\.1|torch.*1\.10\.2.*|torch.*1\.11\.0.*|pytorch-ignite.*0\.4\.10.*|torchaudio.*0\.11\.0.*|torchvision.*0\.12\.0.*|terminado.*0\.15\.0|qudida.*0\.0\.4|expiringdict.*1\.2\.2|botocore.*1\.29\.80|orderedmultidict.*1\.0\.1|deepchecks.*)'
           # psycopg2 is LGPL 2
           # pyzmq is Revised BSD https://github.com/zeromq/pyzmq/blob/main/examples/LICENSE
           # debugpy is MIT https://github.com/microsoft/debugpy/blob/main/LICENSE

backend/deepchecks_monitoring/ee/bgtasks/object_storage_ingestor.py

@@ -28,9 +28,10 @@
 from deepchecks_monitoring.public_models.task import BackgroundWorker, Task
 from deepchecks_monitoring.resources import ResourcesProvider
 from deepchecks_monitoring.schema_models import Model, ModelVersion
-from deepchecks_monitoring.schema_models.column_type import SAMPLE_ID_COL, SAMPLE_TS_COL
+from deepchecks_monitoring.schema_models.column_type import SAMPLE_ID_COL, SAMPLE_TS_COL, ColumnType
 from deepchecks_monitoring.schema_models.data_sources import DataSource
 from deepchecks_monitoring.utils import database
+from deepchecks_monitoring.utils.other import datetime_formatter, string_formatter
 
 __all__ = ['ObjectStorageIngestor']
 
@@ -147,6 +148,12 @@ async def run(self, task: 'Task', session: AsyncSession, resources_provider: Res
                 for prefix in version_prefixes:
                     for df, time in self.ingest_prefix(s3, bucket, f'{version_path}/{prefix}', version.latest_file_time,
                                                        errors, version.model_id, version.id, need_ts=True):
+                        for name, kind in version.features_columns.items():
+                            if kind == ColumnType.CATEGORICAL and name in df.columns:
+                                df[name] = df[name].apply(string_formatter)
+                            elif kind == ColumnType.DATETIME and name in df.columns:
+                                df[name] = df[name].apply(datetime_formatter)
+
                         # For each file, set lock expiry to 360 seconds from now
                         await lock.extend(360, replace_ttl=True)
                         await self.ingestion_backend.log_samples(version, df, session, organization_id, new_scan_time)

backend/deepchecks_monitoring/utils/other.py

@@ -1,17 +1,23 @@
 """Represent global utility functions."""
 import typing as t
+from datetime import datetime
 from ssl import SSLContext
 
+import numpy as np
+import pandas as pd
 import pendulum as pdl
+import rfc3339_validator
 from aiokafka.admin import AIOKafkaAdminClient
 from aiokafka.admin import __version__ as aiokafka_version
 from aiokafka.client import AIOKafkaClient
+from pendulum.datetime import DateTime as PendulumDateTime
 from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
 
 from deepchecks_monitoring.public_models import Organization, User, UserOAuthDTO
 from deepchecks_monitoring.schema_models.model_version import ModelVersion
 
-__all__ = ['generate_random_user', 'generate_test_user', 'datetime_sample_formatter']
+__all__ = ['generate_random_user', 'generate_test_user', 'datetime_sample_formatter',
+           'datetime_formatter', 'string_formatter', 'parse_timestamp']
 
 
 class ExtendedAIOKafkaAdminClient(AIOKafkaAdminClient):  # pylint: disable=missing-class-docstring
@@ -108,3 +114,37 @@ def datetime_sample_formatter(sample: t.Dict, model_version: ModelVersion):
             continue
         if model_columns[col_name].get('format') == 'date-time':
             sample[col_name] = pdl.parse(val)
+
+
+def datetime_formatter(datetime_obj):
+    if datetime_obj is None:
+        return None
+    if isinstance(datetime_obj, pd.Period):
+        datetime_obj = datetime_obj.to_timestamp()
+    elif isinstance(datetime_obj, np.datetime64):
+        datetime_obj = pd.Timestamp(datetime_obj.to_timestamp())
+    return parse_timestamp(datetime_obj).to_iso8601_string()
+
+
+def string_formatter(some_obj):
+    if pd.isna(some_obj):
+        return None
+    return str(some_obj)
+
+
+def parse_timestamp(timestamp: t.Union[int, datetime, str]) -> 'PendulumDateTime':
+    """Parse timestamp to datetime object."""
+    # If no timezone in datetime, assumed to be UTC and converted to local timezone
+    if isinstance(timestamp, int) or np.issubdtype(type(timestamp), np.integer):
+        return pdl.from_timestamp(timestamp, pdl.local_timezone())
+    elif isinstance(timestamp, PendulumDateTime):
+        return timestamp
+    elif isinstance(timestamp, datetime):
+        return pdl.instance(timestamp, pdl.local_timezone())
+    elif isinstance(timestamp, str):
+        if rfc3339_validator.validate_rfc3339(timestamp):
+            return pdl.parse(timestamp)
+        else:
+            raise ValueError(f'Not supported timestamp format for: {timestamp}')
+    else:
+        raise ValueError(f'Not supported timestamp type: {type(timestamp)}')

backend/requirements.txt

@@ -38,4 +38,5 @@ ddtrace>=1.0,<2
 python-json-logger>=2.0.7
 mixpanel==4.10.0
 typing_extensions==4.7.1
-pyarrow==14.0.1
+pyarrow==14.0.1
+rfc3339-validator==0.1.4 # Used to validate date-time format in jsonschema