Merge branch 'feature/feature-store' of github.com:oracle/accelerated-data-science into ODSC-44773/fs_doc_update

Author: KshitizLohia

ads/feature_store/common/spark_session_singleton.py

@@ -9,7 +9,6 @@
 
 from ads.common.decorator.runtime_dependency import OptionalDependency
 import os
-
 from ads.common.oci_client import OCIClientFactory
 
 try:
@@ -32,8 +31,33 @@
     raise
 
 
+def get_env_bool(env_var: str, default: bool = False) -> bool:
+    """
+    :param env_var: Environment variable name
+    :param default: Default environment variable value
+    :return: Value of the boolean env variable
+    """
+    env_val = os.getenv(env_var)
+    if env_val is None:
+        env_val = default
+    else:
+        env_val = env_val.lower()
+        if env_val == "true":
+            env_val = True
+        elif env_val == "false":
+            env_val = False
+        else:
+            raise ValueError(
+                "For environment variable: {0} only string values T/true or F/false are allowed but: \
+                {1} was provided.".format(
+                    env_var, env_val
+                )
+            )
+    return env_val
+
+
 def developer_enabled():
-    return os.getenv("DEVELOPER_MODE")
+    return get_env_bool("DEVELOPER_MODE", False)
 
 
 class SingletonMeta(type):
@@ -75,8 +99,9 @@ def __init__(self, metastore_id: str = None):
                 "spark.hadoop.oracle.dcat.metastore.id", metastore_id
             ).config(
                 "spark.sql.warehouse.dir", metastore.default_managed_table_location
-            )\
-                .config("spark.driver.memory", "16G")
+            ).config(
+                "spark.driver.memory", "16G"
+            )
 
         if developer_enabled():
             # Configure spark session with delta jars only in developer mode. In other cases,

ads/feature_store/common/utils/feature_schema_mapper.py

@@ -241,7 +241,6 @@ def map_feature_type_to_pandas(feature_type):
         raise TypeError(f"Feature Type {feature_type} is not supported for pandas")
 
 
-
 def map_spark_type_to_stats_data_type(spark_type):
     """Maps the spark data types to MLM library data types
     args:

ads/feature_store/common/utils/transformation_query_validator.py

@@ -1,5 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8; -*-
+import copy
+import os
 
 # Copyright (c) 2023 Oracle and/or its affiliates.
 # Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
@@ -41,6 +43,7 @@
 from ads.feature_engineering.feature_type import datetime
 
 logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
 
 
 def get_execution_engine_type(
@@ -117,6 +120,92 @@ def validate_delta_format_parameters(
             raise Exception(f"version number cannot be negative")
 
 
+def show_ingestion_summary(
+    entity_id: str,
+    entity_type: EntityType = EntityType.FEATURE_GROUP,
+    error_details: str = None,
+):
+    """
+    Displays a ingestion summary table with the given entity type and error details.
+
+    Args:
+        entity_id: str
+        entity_type (EntityType, optional): The type of entity being ingested. Defaults to EntityType.FEATURE_GROUP.
+        error_details (str, optional): Details of any errors that occurred during ingestion. Defaults to None.
+    """
+    from tabulate import tabulate
+
+    table_headers = ["entity_id", "entity_type", "ingestion_status", "error_details"]
+    ingestion_status = "Failed" if error_details else "Succeeded"
+
+    table_values = [
+        entity_id,
+        entity_type.value,
+        ingestion_status,
+        error_details if error_details else "None",
+    ]
+
+    logger.info(
+        "Ingestion Summary \n"
+        + tabulate(
+            [table_values],
+            headers=table_headers,
+            tablefmt="fancy_grid",
+            numalign="center",
+            stralign="center",
+        )
+    )
+
+
+def show_validation_summary(ingestion_status: str, validation_output, expectation_type):
+    from tabulate import tabulate
+
+    statistics = validation_output["statistics"]
+
+    table_headers = (
+        ["expectation_type"] + list(statistics.keys()) + ["ingestion_status"]
+    )
+
+    table_values = [expectation_type] + list(statistics.values()) + [ingestion_status]
+
+    logger.info(
+        "Validation Summary \n"
+        + tabulate(
+            [table_values],
+            headers=table_headers,
+            tablefmt="fancy_grid",
+            numalign="center",
+            stralign="center",
+        )
+    )
+
+    rule_table_headers = ["rule_type", "arguments", "status"]
+
+    rule_table_values = [
+        [
+            rule_output["expectation_config"].get("expectation_type"),
+            {
+                key: value
+                for key, value in rule_output["expectation_config"]["kwargs"].items()
+                if key != "batch_id"
+            },
+            rule_output.get("success"),
+        ]
+        for rule_output in validation_output["results"]
+    ]
+
+    logger.info(
+        "Validations Rules Summary \n"
+        + tabulate(
+            rule_table_values,
+            headers=rule_table_headers,
+            tablefmt="fancy_grid",
+            numalign="center",
+            stralign="center",
+        )
+    )
+
+
 def get_features(
     output_columns: List[dict],
     parent_id: str,
@@ -154,8 +243,10 @@ def get_features(
     return features
 
 
-def get_schema_from_pandas_df(df: pd.DataFrame):
-    spark = SparkSessionSingleton().get_spark_session()
+def get_schema_from_pandas_df(df: pd.DataFrame, feature_store_id: str):
+    spark = SparkSessionSingleton(
+        get_metastore_id(feature_store_id)
+    ).get_spark_session()
     converted_df = spark.createDataFrame(df)
     return get_schema_from_spark_df(converted_df)
 
@@ -174,27 +265,29 @@ def get_schema_from_spark_df(df: DataFrame):
     return schema_details
 
 
-def get_schema_from_df(data_frame: Union[DataFrame, pd.DataFrame]) -> List[dict]:
+def get_schema_from_df(
+    data_frame: Union[DataFrame, pd.DataFrame], feature_store_id: str
+) -> List[dict]:
     """
     Given a DataFrame, returns a list of dictionaries that describe its schema.
     If the DataFrame is a pandas DataFrame, it uses pandas methods to get the schema.
     If it's a PySpark DataFrame, it uses PySpark methods to get the schema.
     """
     if isinstance(data_frame, pd.DataFrame):
-        return get_schema_from_pandas_df(data_frame)
+        return get_schema_from_pandas_df(data_frame, feature_store_id)
     else:
         return get_schema_from_spark_df(data_frame)
 
 
 def get_input_features_from_df(
-    data_frame: Union[DataFrame, pd.DataFrame]
+    data_frame: Union[DataFrame, pd.DataFrame], feature_store_id: str
 ) -> List[FeatureDetail]:
     """
     Given a DataFrame, returns a list of FeatureDetail objects that represent its input features.
     Each FeatureDetail object contains information about a single input feature, such as its name, data type, and
     whether it's categorical or numerical.
     """
-    schema_details = get_schema_from_df(data_frame)
+    schema_details = get_schema_from_df(data_frame, feature_store_id)
     feature_details = []
 
     for schema_detail in schema_details:
@@ -263,7 +356,7 @@ def largest_matching_subset_of_primary_keys(left_feature_group, right_feature_gr
 
 
 def convert_pandas_datatype_with_schema(
-        raw_feature_details: List[dict], input_df: pd.DataFrame
+    raw_feature_details: List[dict], input_df: pd.DataFrame
 ) -> pd.DataFrame:
     feature_detail_map = {}
     columns_to_remove = []
@@ -280,21 +373,25 @@ def convert_pandas_datatype_with_schema(
                 .where(pd.notnull(input_df[column]), None)
             )
         else:
-            logger.warning("column" + column + "doesn't exist in the input feature details")
+            logger.warning(
+                "column" + column + "doesn't exist in the input feature details"
+            )
             columns_to_remove.append(column)
-    return input_df.drop(columns = columns_to_remove)
+    return input_df.drop(columns=columns_to_remove)
 
 
 def convert_spark_dataframe_with_schema(
-        raw_feature_details: List[dict], input_df: DataFrame
+    raw_feature_details: List[dict], input_df: DataFrame
 ) -> DataFrame:
     feature_detail_map = {}
     columns_to_remove = []
     for feature_details in raw_feature_details:
         feature_detail_map[feature_details.get("name")] = feature_details
     for column in input_df.columns:
         if column not in feature_detail_map.keys():
-            logger.warning("column" + column + "doesn't exist in the input feature details")
+            logger.warning(
+                "column" + column + "doesn't exist in the input feature details"
+            )
             columns_to_remove.append(column)
 
     return input_df.drop(*columns_to_remove)

ads/feature_store/common/utils/utility.py

@@ -115,7 +115,7 @@ def apply_validations(expectation_details, expectation_suite_name, dataframe):
         str
             A string representation of the validation result.
         """
-        validation_output = None
+        expectation_response = None
         if (
             expectation_details
             and expectation_details.get("expectationType")
@@ -126,14 +126,4 @@ def apply_validations(expectation_details, expectation_suite_name, dataframe):
                 expectation_details, expectation_suite_name, dataframe
             )
 
-            validation_output = str(expectation_response)
-
-            if expectation_details["expectationType"] == ExpectationType.STRICT.value:
-                if not expectation_response["success"]:
-                    raise Exception(
-                        "Expectation failed with statistics: {0} ... Aborting ingestion.".format(
-                            expectation_response["statistics"]
-                        )
-                    )
-
-        return validation_output
+        return expectation_response