DataType Changes,validation output transpose

Author: najiyacl

ads/feature_store/common/enums.py

@@ -295,6 +295,7 @@ class FeatureType(Enum):
     STRING_BINARY_MAP = "STRING_BINARY_MAP"
     STRING_BOOLEAN_MAP = "STRING_BOOLEAN_MAP"
     UNKNOWN = "UNKNOWN"
+    COMPLEX = "COMPLEX"
 
 
 class EntityType(Enum):

ads/feature_store/common/spark_session_singleton.py

@@ -75,7 +75,8 @@ 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")
 
         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

@@ -71,7 +71,7 @@ def map_spark_type_to_feature_type(spark_type):
     if spark_type in spark_type_to_feature_type:
         return spark_type_to_feature_type.get(spark_type)
     else:
-        return FeatureType.UNKNOWN
+        return FeatureType.COMPLEX
 
 
 def map_pandas_type_to_feature_type(feature_name, values):
@@ -180,7 +180,7 @@ def map_feature_type_to_spark_type(feature_type):
     if feature_type_in in spark_types:
         return spark_types.get(feature_type_in)
     else:
-        return "UNKNOWN"
+        return "COMPLEX"
 
 
 def get_raw_data_source_schema(raw_feature_details: List[dict]):
@@ -225,30 +225,22 @@ def map_feature_type_to_pandas(feature_type):
         FeatureType.INTEGER: "int32",
         FeatureType.DECIMAL: "object",
         FeatureType.DATE: "object",
+        FeatureType.STRING_ARRAY: "object",
+        FeatureType.INTEGER_ARRAY: "object",
+        FeatureType.LONG_ARRAY: "object",
+        FeatureType.FLOAT_ARRAY: "object",
+        FeatureType.DOUBLE_ARRAY: "object",
+        FeatureType.TIMESTAMP_ARRAY: "object",
+        FeatureType.BOOLEAN_ARRAY: "object",
+        # FeatureType.DECIMAL_ARRAY: "object",
+        FeatureType.DATE_ARRAY: "object",
     }
     if feature_type_in in supported_feature_type:
         return supported_feature_type.get(feature_type_in)
     else:
         raise TypeError(f"Feature Type {feature_type} is not supported for pandas")
 
 
-def convert_pandas_datatype_with_schema(
-    raw_feature_details: List[dict], input_df: pd.DataFrame
-):
-    feature_detail_map = {}
-    for feature_details in raw_feature_details:
-        feature_detail_map[feature_details.get("name")] = feature_details
-    for column in input_df.columns:
-        if column in feature_detail_map.keys():
-            feature_details = feature_detail_map[column]
-            feature_type = feature_details.get("featureType")
-            pandas_type = map_feature_type_to_pandas(feature_type)
-            input_df[column] = (
-                input_df[column]
-                .astype(pandas_type)
-                .where(pd.notnull(input_df[column]), None)
-            )
-
 
 def map_spark_type_to_stats_data_type(spark_type):
     """Maps the spark data types to MLM library data types

ads/feature_store/common/utils/utility.py

@@ -11,11 +11,12 @@
 from ads.common.decorator.runtime_dependency import OptionalDependency
 from ads.feature_store.common.utils.feature_schema_mapper import (
     map_spark_type_to_feature_type,
-    map_pandas_type_to_feature_type,
+    map_feature_type_to_pandas,
 )
 from ads.feature_store.feature import Feature, DatasetFeature
 from ads.feature_store.feature_group_expectation import Rule, Expectation
 from ads.feature_store.input_feature_detail import FeatureDetail
+from ads.feature_store.common.spark_session_singleton import SparkSessionSingleton
 
 try:
     from pyspark.pandas import DataFrame
@@ -154,18 +155,9 @@ def get_features(
 
 
 def get_schema_from_pandas_df(df: pd.DataFrame):
-    schema_details = []
-
-    for order_number, field in enumerate(df.columns, start=1):
-        details = {
-            "name": field,
-            "feature_type": map_pandas_type_to_feature_type(field, df[field]),
-            "order_number": order_number,
-        }
-
-        schema_details.append(details)
-
-    return schema_details
+    spark = SparkSessionSingleton().get_spark_session()
+    converted_df = spark.createDataFrame(df)
+    return get_schema_from_spark_df(converted_df)
 
 
 def get_schema_from_spark_df(df: DataFrame):
@@ -268,3 +260,45 @@ def largest_matching_subset_of_primary_keys(left_feature_group, right_feature_gr
     common_keys = left_primary_keys.intersection(right_primary_keys)
 
     return common_keys
+
+
+def convert_pandas_datatype_with_schema(
+        raw_feature_details: List[dict], input_df: pd.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 in feature_detail_map.keys():
+            feature_details = feature_detail_map[column]
+            feature_type = feature_details.get("featureType")
+            pandas_type = map_feature_type_to_pandas(feature_type)
+            input_df[column] = (
+                input_df[column]
+                .astype(pandas_type)
+                .where(pd.notnull(input_df[column]), None)
+            )
+        else:
+            logger.warning("column" + column + "doesnt exist in the input feature details")
+            columns_to_remove.append(column)
+    return input_df.drop(columns = columns_to_remove)
+
+
+def validate_spark_dataframe_schema(raw_feature_details: List[dict], input_df: 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 + "doesnt exist in the input feature details")
+            columns_to_remove.append(column)
+
+    return input_df.drop(*columns_to_remove)
+
+
+def validate_input_feature_details(input_feature_details, data_frame):
+    if isinstance(data_frame, pd.DataFrame):
+        return convert_pandas_datatype_with_schema(input_feature_details, data_frame)
+    return validate_spark_dataframe_schema(input_feature_details, data_frame)

ads/feature_store/docs/source/feature_group.rst

@@ -175,10 +175,6 @@ Statistics Results
 ==================
 You can call the ``get_statistics()`` method of the FeatureGroup instance to fetch validation results for a specific ingestion job.
 
-.. note::
-
-  PyDeequ is a Python API for Deequ, a library built on top of Apache Spark for defining "unit tests for data", which measure data quality in large datasets.
-
 .. code-block:: python3
 
   # Fetch stats results for a feature group job
@@ -196,26 +192,17 @@ With a FeatureGroup instance, we can get the last feature group job details usin
 
   # Fetch validation results for a feature group
   feature_group_job = feature_group.get_last_job()
-  df = feature_group_job.get_validation().to_pandas()
-  df.show()
+  feature_group_job.get_validation_output_df()
 
 Get features
 =============
-You can call the ``get_features_dataframe()`` method of the FeatureGroup instance to fetch features in a feature group
+You can call the ``get_features_df`` method of the FeatureGroup instance to fetch features in a feature group
 
 .. code-block:: python3
 
   # Fetch features for a feature group
-  df = feature_group.get_features_dataframe()
-
-Get input schema details
-==========================
-You can call the ``get_input_schema_dataframe()`` method of the FeatureGroup instance to fetch input schema details of a feature group
-
-.. code-block:: python3
+  df = feature_group.get_features_df()
 
-  # Fetch features for a feature group
-  df = feature_group.get_input_schema_dataframe()
 
 Filter
 ======
@@ -363,31 +350,31 @@ The data will be stored in a data type native to each store. There is an option
          - STRING
          - Textual data
        * - ArrayType(IntegerType())
-         - object (list), object (np.ndarray) - not supported
+         - object (list), object (np.ndarray)
          - INTEGER_ARRAY
          - List of values
        * - ArrayType(LongType())
-         - object (list), object (np.ndarray) - not supported
+         - object (list), object (np.ndarray)
          - LONG_ARRAY
          - List of values
        * - ArrayType(FloatType())
-         - object (list), object (np.ndarray) - not supported
+         - object (list), object (np.ndarray)
          - FLOAT_ARRAY
          - List of values
        * - ArrayType(DoubleType())
-         - object (list), object (np.ndarray) - not supported
+         - object (list), object (np.ndarray)
          - DOUBLE_ARRAY
          - List of values
        * - ArrayType(BinaryType())
-         - object (list), object (np.ndarray) - not supported
+         - object (list), object (np.ndarray)
          - BINARY_ARRAY
          - List of values
        * - ArrayType(DateType())
-         - object (list), object (np.ndarray) - not supported
+         - object (list), object (np.ndarray)
          - DATE_ARRAY
          - List of values
        * - ArrayType(TimestampType())
-         - object (list), object (np.ndarray) - not supported
+         - object (list), object (np.ndarray)
          - TIMESTAMP_ARRAY
          - List of values
        * - StructType

ads/feature_store/execution_strategy/spark/spark_execution.py

@@ -27,9 +27,6 @@
     EntityType,
 )
 from ads.feature_store.common.spark_session_singleton import SparkSessionSingleton
-from ads.feature_store.common.utils.feature_schema_mapper import (
-    convert_pandas_datatype_with_schema,
-)
 from ads.feature_store.common.utils.transformation_utils import TransformationUtils
 from ads.feature_store.data_validation.great_expectation import ExpectationService
 from ads.feature_store.dataset_job import DatasetJob
@@ -41,6 +38,9 @@
 from ads.feature_store.transformation import Transformation
 
 from ads.feature_store.feature_statistics.statistics_service import StatisticsService
+from ads.feature_store.common.utils.utility import (
+    validate_input_feature_details
+)
 
 logger = logging.getLogger(__name__)
 
@@ -177,11 +177,8 @@ def _save_offline_dataframe(
             database = feature_group.entity_id
             self.spark_engine.create_database(database)
 
-            if isinstance(data_frame, pd.DataFrame):
-                if not feature_group.is_infer_schema:
-                    convert_pandas_datatype_with_schema(
-                        feature_group.input_feature_details, data_frame
-                    )
+            if not feature_group.is_infer_schema:
+                data_frame = validate_input_feature_details(feature_group.input_feature_details, data_frame)
 
             # TODO: Get event timestamp column and apply filtering basis from and to timestamp
 

ads/feature_store/feature_group_job.py

@@ -177,7 +177,7 @@ def get_validation_output_df(self) -> "pandas.DataFrame":
         )
 
         # Convert Python object to Pandas DataFrame
-        validation_output_df = pandas.json_normalize(validation_output_json)
+        validation_output_df = pandas.json_normalize(validation_output_json).transpose()
 
         # return the validation output DataFrame
         return validation_output_df

ads/feature_store/validation_output.py

@@ -22,7 +22,7 @@ def to_pandas(self) -> pd.DataFrame:
             The validation output information as a pandas DataFrame.
         """
         if self.content:
-            profile_result = pd.json_normalize(self.content)
+            profile_result = pd.json_normalize(self.content).transpose()
             return profile_result
 
     @property

tests/integration/feature_store/test_dataset_validations.py

@@ -55,7 +55,7 @@ def test_dataset_validation_operations(self):
         assert dataset.oci_dataset.id
 
         dataset.materialise()
-        df = dataset.get_validation_output().to_pandas()
+        df = dataset.get_validation_output().to_pandas().T
         assert df is not None
         assert "success" in df.columns
         assert True in df["success"].values