Merge pull request #136 from splicemachine/DBAAS-5219

Dbaas 5219 - Pipeline feature aggregation support

Author: Ben Epstein

requirements-docs.txt

@@ -1,8 +1,5 @@
 py4j==0.10.7.0
-pytest
 mlflow==1.8.0
-pyyaml==5.4
-mleap==0.15.0
 graphviz==0.13
 requests
 gorilla==0.3.0
@@ -11,9 +8,7 @@ pyspark-dist-explore==0.1.8
 numpy==1.18.2
 pandas==1.0.3
 scipy==1.4.1
-tensorflow==2.2.1
 pyspark
-h2o-pysparkling-2.4==3.28.1.2-1
+h2o-pysparkling-3.0==3.32.0.4-1
 sphinx-tabs
 IPython
-cloudpickle==1.6.0

splicemachine/features/feature_store.py

@@ -0,0 +1,2 @@
+from .utils import AggWindow, FeatureAgg
+from .feature_aggregation import FeatureAggregation

splicemachine/features/pipelines/__init__.py

@@ -0,0 +1,37 @@
+from typing import List
+
+class FeatureAggregation:
+    def __init__(self, column_name: str, agg_functions: List[str], agg_windows: List[str],
+                       feature_name_prefix: str = None, agg_default_value: float = None ):
+        """
+        A class abstraction for defining aggregations that will generate a set of features for a feature set from a Pipeline
+
+        :param column_name: Name of the source column from a Source SQL statement
+        :param agg_function: type of SQL aggregation (sum, min, max, avg, count etc). Available aggregations
+        are in the splicemachine.features.pipelines.FeatureAgg class
+        :param agg_windows: list of time windows over which to perform the aggregation ("1w","5d","3m")
+        available windows: "s"econd, "m"inute, "d"ay, "w"eek, "mn"onth, "q"uarter, "y"ear
+        :param agg_default_value: the default value in case of a null (no activity in the window)
+        available windows: "s"econd, "m"ute, "d"ay, "w"eek, "mn"onth, "q"uarter, "y"ear
+        setting this parameter would create 1 new feature for every agg_window+agg_function pair specified
+
+        The name of the features created through this FeatureAggregation will be: schema_table_sourcecol_aggfunc_aggwindow
+
+        :Example:
+            .. code-block:: python
+
+                from splicemachine.features.pipelines import AggWindow, FeatureAgg, FeatureAggregation
+                FeatureAggregation('revenue', [FeatureAgg.SUM, FeatureAgg.AVG],\
+                        [AggWindow.get_window(5, AggWindow.DAY), AggWindow.get_window(10, AggWindow.SECOND)], 0.0)
+
+        would yield:
+            customer_rfm_revenue_wrate_sum_1d
+            customer_rfm_revenue_wrate_sum_5w
+            customer_rfm_revenue_wrate_mean_1d
+            customer_rfm_revenue_wrate_mean_5w
+      """
+        self.column_name=column_name
+        self.agg_functions= agg_functions
+        self.agg_windows = agg_windows
+        self.agg_default_value = agg_default_value
+        self.feature_name_prefix = feature_name_prefix

splicemachine/features/pipelines/feature_aggregation.py

@@ -0,0 +1,54 @@
+class FeatureAgg:
+    """
+    A class defining the valid Aggregation functions available to features in order to create FeatureAggregations
+    """
+    SUM = 'sum'
+    COUNT = 'count'
+    AVG = 'avg'
+    MEAN = 'avg'
+    MIN = 'min'
+    MAX = 'max'
+
+    @staticmethod
+    def get_valid():
+        return (FeatureAgg.SUM, FeatureAgg.COUNT, FeatureAgg.AVG, FeatureAgg.MIN, FeatureAgg.MAX)
+
+class AggWindow:
+    """
+    A class defining the valid window types available to aggregation functions for use in FeatureAggregations
+    """
+    SECOND = 's'
+    MINUTE = 'm'
+    HOUR = 'h'
+    DAY = 'd'
+    WEEK = 'w'
+    MONTH = 'mn'
+    QUARTER = 'q'
+    YEAR = 'y'
+
+    @staticmethod
+    def get_valid():
+        return (AggWindow.SECOND, AggWindow.MINUTE, AggWindow.HOUR, AggWindow.DAY,
+                AggWindow.WEEK, AggWindow.MONTH, AggWindow.QUARTER, AggWindow.YEAR)
+
+    @staticmethod
+    def get_window(length: int, window: str):
+        """
+        A function to help in creating a valid window aggregation. Validates and returns the proper window aggregation
+        syntax for use in FeatureAggregations
+
+        :param length: The length of time for the window
+        :param window: The window type as defined in WindowAgg.get_valid()
+        :return: (str) the proper window aggregation syntax
+
+        :Example:
+            .. code-block:: python
+
+
+                WindowAgg.get_window(5, WindowAgg.SECOND) -> '5s'
+                WindowAgg.get_window(10, WindowAgg.MONTH) -> '10mn'
+        """
+        assert window in AggWindow.get_valid(), f'The provided window {window} is not valid. ' \
+                                                f'Use one of {AggWindow.get_valid()}'
+        assert length > 0 and type(length) == int, f'Length must be a positive, nonzero integer, but got {length}'
+        return f'{length}{window}'

splicemachine/features/pipelines/utils.py

@@ -30,8 +30,9 @@ class Endpoints:
     """
     DEPLOYMENTS: str = "deployments"
     FEATURES: str = "features"
+    FEATURE_DETAILS: str = "feature-details"
     FEATURE_SETS: str = "feature-sets"
-    FEATURE_SET_DESCRIPTIONS: str = "feature-set-descriptions"
+    FEATURE_SET_DETAILS: str = "feature-set-details"
     DEPLOY_FEATURE_SET: str = "deploy-feature-set"
     FEATURE_VECTOR: str = "feature-vector"
     FEATURE_VECTOR_SQL: str = "feature-vector-sql"
@@ -40,10 +41,15 @@ class Endpoints:
     TRAINING_SET_FROM_DEPLOYMENT: str = "training-set-from-deployment"
     TRAINING_SET_FROM_VIEW: str = "training-set-from-view"
     TRAINING_VIEWS: str = "training-views"
-    TRAINING_VIEW_DESCRIPTIONS: str = "training-view-descriptions"
+    TRAINING_VIEW_DETAILS: str = "training-view-details"
     TRAINING_VIEW_FEATURES: str = "training-view-features"
     TRAINING_VIEW_ID: str = "training-view-id"
     SUMMARY: str = "summary"
+    SOURCE: str = "source"
+    AGG_FEATURE_SET_FROM_SOURCE: str = 'agg-feature-set-from-source'
+    BACKFILL_SQL: str = 'backfill-sql'
+    BACKFILL_INTERVALS: str = 'backfill-intervals'
+    PIPELINE_SQL: str = 'pipeline-sql'
 
 def make_request(url: str, endpoint: str, method: str, auth: HTTPBasicAuth,
                  params: Optional[Dict[str, Any]] = None,