DataType Changes, validation output transpose (#246)

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,47 @@ 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
+) -> 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 + "doesn't exist in the input feature details")
+            columns_to_remove.append(column)
+    return input_df.drop(columns = columns_to_remove)
+
+
+def convert_spark_dataframe_with_schema(
+        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")
+            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 convert_spark_dataframe_with_schema(input_feature_details, data_frame)

ads/feature_store/docs/source/feature_group.rst

@@ -152,32 +152,49 @@ Feature store provides an API similar to Pandas to join feature groups together
 
 Save expectation entity
 =======================
-With a ``FeatureGroup`` instance, we can save the expectation entity using ``save_expectation()``
+With a ``FeatureGroup`` instance, You can save the expectation details using ``with_expectation_suite()`` with parameters
+
+- ``expectation_suite: ExpectationSuite``. ExpectationSuit of great expectation
+- ``expectation_type: ExpectationType``. Type of expectation
+        - ``ExpectationType.STRICT``: Fail the job if expectation not met
+        - ``ExpectationType.LENIENT``: Pass the job even if expectation not met
 
 .. note::
 
   Great Expectations is a Python-based open-source library for validating, documenting, and profiling your data. It helps you to maintain data quality and improve communication about data between teams. Software developers have long known that automated testing is essential for managing complex codebases.
 
 .. image:: figures/validation.png
 
-The ``.save_expectation()`` method takes the following optional parameter:
+.. code-block:: python3
 
-- ``expectation: Expectation``. Expectation of great expectation
-- ``expectation_type: ExpectationType``. Type of expectation
-        - ``ExpectationType.STRICT``: Fail the job if expectation not met
-        - ``ExpectationType.LENIENT``: Pass the job even if expectation not met
+    expectation_suite = ExpectationSuite(
+        expectation_suite_name="expectation_suite_name"
+    )
+    expectation_suite.add_expectation(
+        ExpectationConfiguration(
+            expectation_type="expect_column_values_to_not_be_null",
+            kwargs={"column": "<column>"},
+        )
 
-.. code-block:: python3
+    feature_group_resource = (
+        FeatureGroup()
+        .with_feature_store_id(feature_store.id)
+        .with_primary_keys(["<key>"])
+        .with_name("<name>")
+        .with_entity_id(entity.id)
+        .with_compartment_id(<compartment_id>)
+        .with_schema_details_from_dataframe(<datframe>)
+        .with_expectation_suite(
+            expectation_suite=expectation_suite,
+            expectation_type=ExpectationType.STRICT,
+         )
+    )
 
-  feature_group.save_expectation(expectation_suite, expectation_type="STRICT")
+You can call the ``get_validation_output()`` method of the FeatureGroup instance to fetch validation results for a specific ingestion job.
 
 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.
+You can call the ``get_statistics()`` method of the FeatureGroup instance to fetch statistics for a specific ingestion job.
 
 .. code-block:: python3
 
@@ -196,26 +213,16 @@ 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()
 
 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
+  df = feature_group.get_features_df()
 
-.. code-block:: python3
-
-  # Fetch features for a feature group
-  df = feature_group.get_input_schema_dataframe()
 
 Filter
 ======
@@ -308,7 +315,8 @@ The data will be stored in a data type native to each store. There is an option
 
     Offline data types
     ###################
-    Please refer to the following mapping when registering a Spark DataFrame, or a Pandas DataFrame.
+    Please refer to the following mapping when registering a Spark DataFrame, or a Pandas DataFrame.For spark dataframes we support
+    all the data types and the ones which are not specified in the following table will be mapped to  Offline Feature Type COMPLEX
 
     .. list-table::
        :widths: 20 25 25 40
@@ -363,31 +371,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
          - 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/docs/source/release_notes.rst

@@ -3,6 +3,33 @@
 =============
 Release Notes
 =============
+1.1
+---
+
+.. note::
+
+    .. list-table::
+      :header-rows: 1
+
+      * - Package Name
+        - Latest Version
+        - Notes
+      * - Conda pack
+        - `https://objectstorage.us-ashburn-1.oraclecloud.com/n/bigdatadatasciencelarge/b/service-conda-packs-fs/o/service_pack/cpu/PySpark_3.2_and_Feature_Store/1.0/fspyspark32_p38_cpu_v1#conda`
+        -
+      * - SERVICE_VERSION
+        - 0.1.212.master
+        -
+      * - Terraform Stack
+        - `link <https://objectstorage.us-ashburn-1.oraclecloud.com/p/vZogtXWwHqbkGLeqyKiqBmVxdbR4MK4nyOBqDsJNVE4sHGUY5KFi4T3mOFGA3FOy/n/idogsu2ylimg/b/oci-feature-store/o/beta/terraform/feature-store-terraform.zip>`__
+        - Par link expires Jan 5, 2026
+
+
+Release notes: July 5, 2023
+
+* [FEATURE] Supporting Offline Feature Type COMPLEX
+* [[DOCS] Data Type update for Offline Feature Type COMPLEX
+
 1.0
 ---
 
@@ -21,10 +48,9 @@ Release Notes
         - 0.1.209.master
         -
       * - Terraform Stack
-        - `link <https://objectstorage.us-ashburn-1.oraclecloud.com/p/MokUwWRFZLj1Dgfev7D_0ALc8YL7jEBmM8FIeWH3AysnH2zxavptSobzR6ezErfZ/n/idogsu2ylimg/b/oci-feature-store/o/beta/terraform/feature-store-terraform.zip>`__
+        - `link <https://objectstorage.us-ashburn-1.oraclecloud.com/p/vZogtXWwHqbkGLeqyKiqBmVxdbR4MK4nyOBqDsJNVE4sHGUY5KFi4T3mOFGA3FOy/n/idogsu2ylimg/b/oci-feature-store/o/beta/terraform/feature-store-terraform.zip>`__
         - Par link expires Jan 5, 2026
 
-
 Release notes: June 15, 2023
 
 * [FEATURE] Included ``FeatureStore``, ``FeatureGroup``, ``Dataset``, ``Entity`` and ``Transformation`` concepts for feature store.

ads/feature_store/docs/source/terraform.rst

@@ -30,21 +30,21 @@ Feature Store users need to provide the following access permissions in order to
 
     define tenancy <feature store service tenancy> as <feature store service tenancy ocid>
     endorse group <feature store user group> to read repos in tenancy <feature store service tenancy>
-    allow group <feature store user group> to manage orm-stacks in compartment <compartmentId>
-    allow group <feature store user group> to manage orm-jobs in compartment <compartmentId>
-    allow group <feature store user group> to manage object-family in compartment <compartmentId>
-    allow group <feature store user group> to manage users in compartment <compartmentId>
-    allow group <feature store user group> to manage instance-family in compartment <compartmentId>
-    allow group <feature store user group> to manage tag-namespaces in compartment <compartmentId>
-    allow group <feature store user group> to manage groups in compartment <compartmentId>
-    allow group <feature store user group> to manage policies in compartment <compartmentId>
-    allow group <feature store user group> to manage dynamic-groups in compartment <compartmentId>
-    allow group <feature store user group> to manage virtual-network-family in compartment <compartmentId>
-    allow group <feature store user group> to manage functions-family in compartment <compartmentId>
-    allow group <feature store user group> to inspect compartments in compartment <compartmentId>
-    allow group <feature store user group> to manage cluster-family in compartment <compartmentId>
-    allow group <feature store user group> to manage mysql-family in compartment <compartmentId>
-    allow group <feature store user group> to manage api-gateway-family in compartment <compartmentId>
+    allow group <feature store user group> to manage orm-stacks in compartment <compartmentName>
+    allow group <feature store user group> to manage orm-jobs in compartment <compartmentName>
+    allow group <feature store user group> to manage object-family in compartment <compartmentName>
+    allow group <feature store user group> to manage users in compartment <compartmentName>
+    allow group <feature store user group> to manage instance-family in compartment <compartmentName>
+    allow group <feature store user group> to manage tag-namespaces in compartment <compartmentName>
+    allow group <feature store user group> to manage groups in compartment <compartmentName>
+    allow group <feature store user group> to manage policies in compartment <compartmentName>
+    allow group <feature store user group> to manage dynamic-groups in compartment <compartmentName>
+    allow group <feature store user group> to manage virtual-network-family in compartment <compartmentName>
+    allow group <feature store user group> to manage functions-family in compartment <compartmentName>
+    allow group <feature store user group> to inspect compartments in compartment <compartmentName>
+    allow group <feature store user group> to manage cluster-family in compartment <compartmentName>
+    allow group <feature store user group> to manage mysql-family in compartment <compartmentName>
+    allow group <feature store user group> to manage api-gateway-family in compartment <compartmentName>
 
 Deploy Using Oracle Resource Manager
 ====================================