Merge pull request #36 from AbsaOSS/Release/2.0.8

Release/2.0.8

Author: MDobransky

CHANGELOG.md

@@ -1,6 +1,12 @@
 # Change Log
 All notable changes to this project will be documented in this file.
 
+## 2.0.8 - 2025-05-xx
+  ### Runner
+  - expanded config overrides logging
+  - added config validation
+  - added owner attribute to GroupMetadata class
+
 ## 2.0.7 - 2025-04-15
   ### Runner
   - email reporting is now optional

README.md

@@ -1,4 +1,3 @@
-from pydantic import BaseModelfrom rialto.runner.config_loader import PipelineConfigfrom rialto.jobs import config
 
 # Rialto
 
@@ -106,7 +105,6 @@ pipelines: # a list of pipelines to run
   metadata_manager: # optional
       metadata_schema: catalog.metadata # schema where metadata is stored
   feature_loader: # optional
-      config_path: model_features_config.yaml # path to the feature loader configuration file
       feature_schema: catalog.feature_tables # schema where feature tables are stored
       metadata_schema: catalog.metadata # schema where metadata is stored
   extras: #optional arguments processed as dictionary
@@ -171,23 +169,21 @@ overrides={"pipelines[name=SimpleGroup].target.target_schema": "new_schema"},
 #### Injecting/Replacing whole sections
 You can directly replace a bigger section of the configuration by providing a dictionary
 When the whole section doesn't exist, it will be added to the configuration, however it needs to be added as a whole.
-i.e. if the yaml file doesn't specify feature_loader, you can't just add a feature_loader.config_path, you need to add the whole section.
+i.e. if the yaml file doesn't specify feature_loader, you can't just add a feature_loader.feature_schema, you need to add the whole section.
 ```python
 overrides={"pipelines[name=SimpleGroup].feature_loader":
-                           {"config_path": "features_cfg.yaml",
-                            "feature_schema": "catalog.features",
+                           {"feature_schema": "catalog.features",
                             "metadata_schema": "catalog.metadata"}}
 ```
 
 #### Multiple overrides
 You can provide multiple overrides at once, the order of execution is not guaranteed
 ```python
-overrides={"runner.watch_period_value": 4,
-           "runner.watch_period_units": "weeks",
+overrides={"runner.watched_period_value": 4,
+           "runner.watched_period_units": "weeks",
            "pipelines[name=SimpleGroup].target.target_schema": "new_schema",
            "pipelines[name=SimpleGroup].feature_loader":
-                           {"config_path": "features_cfg.yaml",
-                            "feature_schema": "catalog.features",
+                           {"feature_schema": "catalog.features",
                             "metadata_schema": "catalog.metadata"}
            }
 ```
@@ -630,6 +626,7 @@ GroupMetadata
     frequency: Schedule # generation frequency
     description: str # group description
     key: List[str] # group primary keys
+    owner: str # owner of the group data (table)
     fs_name: str = None # actual table name of this feature group in DataBricks
     features: List[str] = None # A list of feature names belonging to this group
 ```

docs/source/conf.py

@@ -28,7 +28,7 @@
 project = "rialto"
 copyright = "2022, Marek Dobransky"
 author = "Marek Dobransky"
-release = "2.0.7"
+release = "2.0.8"
 
 # -- General configuration ---------------------------------------------------
 # https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration

pyproject.toml

@@ -1,7 +1,7 @@
 [tool.poetry]
 name = "rialto"
 
-version = "2.0.7"
+version = "2.0.8"
 
 packages = [
     { include = "rialto" },
@@ -54,7 +54,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.black]
 line-length = 120
-target-version = ["py36"]
+target-version = ["py310"]
 
 [tool.isort]
 profile = "black"

rialto/metadata/data_classes/group_metadata.py

@@ -32,6 +32,7 @@ class GroupMetadata:
     frequency: Schedule
     description: str
     key: List[str]
+    owner: str
     fs_name: str = None
     features: List[str] = None
 
@@ -42,7 +43,7 @@ def __repr__(self) -> str:
             f"name={self.name!r}, frequency={self.frequency!r}, "
             f"feature store name={self.fs_name!r},"
             f"description={self.description!r}, key={self.key!r}, "
-            f"features={self.features!r}"
+            f"features={self.features!r}, owner={self.owner!r}"
             ")"
         )
 
@@ -65,7 +66,7 @@ def to_tuple(self) -> Tuple:
         """
         if not self.fs_name:
             self.fs_name = class_to_catalog_name(self.name)
-        return (self.name, self.frequency.value, self.description, self.key, self.fs_name)
+        return (self.name, self.frequency.value, self.description, self.key, self.fs_name, self.owner)
 
     @classmethod
     def from_spark(cls, schema: Row) -> Self:
@@ -81,4 +82,5 @@ def from_spark(cls, schema: Row) -> Self:
             frequency=Schedule[schema.group_frequency],
             description=schema.group_description,
             key=schema.group_key,
+            owner=schema.group_owner
         )

rialto/runner/config_loader.py

@@ -18,65 +18,69 @@
 
 from typing import Dict, List, Optional
 
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict
 
 from rialto.common.utils import load_yaml
 from rialto.runner.config_overrides import override_config
 
 
-class IntervalConfig(BaseModel):
+class BaseConfig(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+
+class IntervalConfig(BaseConfig):
     units: str
     value: int
 
 
-class ScheduleConfig(BaseModel):
+class ScheduleConfig(BaseConfig):
     frequency: str
     day: Optional[int] = 0
     info_date_shift: Optional[List[IntervalConfig]] = IntervalConfig(units="days", value=0)
 
 
-class DependencyConfig(BaseModel):
+class DependencyConfig(BaseConfig):
     table: str
     name: Optional[str] = None
     date_col: str
     interval: IntervalConfig
 
 
-class ModuleConfig(BaseModel):
+class ModuleConfig(BaseConfig):
     python_module: str
     python_class: str
 
 
-class MailConfig(BaseModel):
+class MailConfig(BaseConfig):
     sender: str
     to: List[str]
     smtp: str
     subject: str
     sent_empty: Optional[bool] = False
 
 
-class RunnerConfig(BaseModel):
+class RunnerConfig(BaseConfig):
     watched_period_units: str
     watched_period_value: int
     mail: Optional[MailConfig] = None
     bookkeeping: Optional[str] = None
 
 
-class TargetConfig(BaseModel):
+class TargetConfig(BaseConfig):
     target_schema: str
     target_partition_column: str
 
 
-class MetadataManagerConfig(BaseModel):
+class MetadataManagerConfig(BaseConfig):
     metadata_schema: str
 
 
-class FeatureLoaderConfig(BaseModel):
+class FeatureLoaderConfig(BaseConfig):
     feature_schema: str
     metadata_schema: str
 
 
-class PipelineConfig(BaseModel):
+class PipelineConfig(BaseConfig):
     name: str
     module: ModuleConfig
     schedule: ScheduleConfig
@@ -87,7 +91,7 @@ class PipelineConfig(BaseModel):
     extras: Optional[Dict] = {}
 
 
-class PipelinesConfig(BaseModel):
+class PipelinesConfig(BaseConfig):
     runner: RunnerConfig
     pipelines: list[PipelineConfig]
 

rialto/runner/config_overrides.py

@@ -35,7 +35,7 @@ def _override(config, path, value) -> Dict:
     if "[" in key:
         name, index = _split_index_key(key)
         if name not in config:
-            raise ValueError(f"Invalid key {name}")
+            raise ValueError(f"Invalid key: {name}")
         if "=" in index:
             index = _find_first_match(config[name], index)
         else:
@@ -54,10 +54,12 @@ def _override(config, path, value) -> Dict:
             raise IndexError(f"Index {index} out of bounds for key {key}")
     else:
         if len(path) == 1:
+            if key not in config:
+                logger.warning(f"Adding new key: {key} with value {value}")
             config[key] = value
         else:
             if key not in config:
-                raise ValueError(f"Invalid key {key}")
+                raise ValueError(f"Invalid key: {key}")
             config[key] = _override(config[key], path[1:], value)
     return config
 
@@ -70,7 +72,7 @@ def override_config(config: Dict, overrides: Dict) -> Dict:
     :return: Overridden config
     """
     for path, value in overrides.items():
-        logger.info("Applying override: ", path, value)
+        logger.info(f"Applying override:\npath: {path}\nvalue: {value}")
         config = _override(config, path.split("."), value)
 
     return config

rialto/runner/runner.py

@@ -258,7 +258,7 @@ def _run_pipeline(self, pipeline: PipelineConfig):
                 )
             except Exception as error:
                 logger.error(f"An exception occurred in pipeline {pipeline.name}")
-                logger.error(error)
+                logger.exception(error)
                 self.tracker.add(
                     Record(
                         job=pipeline.name,

tests/metadata/resources.py

@@ -22,6 +22,7 @@
         StructField("group_description", StringType(), False),
         StructField("group_key", ArrayType(StringType(), True), False),
         StructField("group_fs_name", StringType(), False),
+        StructField("group_owner", StringType(), False),
     ]
 )
 
@@ -35,8 +36,8 @@
 )
 
 group_base = [
-    ("Group1", "weekly", "group1", ["key1"], "group_1"),
-    ("Group2", "monthly", "group2", ["key2", "key3"], "group_2"),
+    ("Group1", "weekly", "group1", ["key1"], "group_1", "owner_1"),
+    ("Group2", "monthly", "group2", ["key2", "key3"], "group_2", "owner_2"),
 ]
 
 feature_base = [
@@ -50,6 +51,7 @@
     frequency=Schedule.weekly,
     description="group1",
     key=["key1"],
+    owner="owner_1",
 )
 
 group_md2 = GroupMetadata(
@@ -58,6 +60,7 @@
     frequency=Schedule.monthly,
     description="group2",
     key=["key2", "key3"],
+    owner="owner_2",
     features=["Feature1", "Feature2"],
 )
 

tests/runner/overrider.yaml

@@ -50,7 +50,6 @@ pipelines:
       target_schema: catalog.schema
       target_partition_column: "INFORMATION_DATE"
     feature_loader:
-      config_path: path/to/config.yaml
       feature_schema: catalog.feature_tables
       metadata_schema: catalog.metadata
     metadata_manager: