Merge pull request #20 from codeflare/lineage_semantics

Resolved conflicts and merging lineage semantics into develop.

Author: chcost

codeflare/pipelines/Datamodel.py

@@ -1,6 +1,11 @@
-from sklearn.base import BaseEstimator
 from abc import ABC, abstractmethod
+import uuid
+from enum import Enum
+
 
+import sklearn.base as base
+from sklearn.base import TransformerMixin
+from sklearn.base import BaseEstimator
 
 class Xy:
     """
@@ -35,9 +40,12 @@ class XYRef:
     computed), these holders are essential to the pipeline constructs.
     """
 
-    def __init__(self, Xref, yref):
+    def __init__(self, Xref, yref, prev_noderef=None, curr_noderef=None, prev_Xyrefs = None):
         self.__Xref__ = Xref
         self.__yref__ = yref
+        self.__prevnoderef__ = prev_noderef
+        self.__currnoderef__ = curr_noderef
+        self.__prev_Xyrefs__ = prev_Xyrefs
 
     def get_Xref(self):
         """
@@ -51,6 +59,32 @@ def get_yref(self):
         """
         return self.__yref__
 
+    def get_prevnoderef(self):
+        return self.__prevnoderef__
+
+    def get_currnoderef(self):
+        return self.__currnoderef__
+
+    def get_prev_xyrefs(self):
+        return self.__prev_Xyrefs__
+
+
+class NodeInputType(Enum):
+    OR = 0,
+    AND = 1
+
+
+class NodeFiringType(Enum):
+    ANY = 0,
+    ALL = 1
+
+
+class NodeStateType(Enum):
+    STATELESS = 0,
+    IMMUTABLE = 1,
+    MUTABLE_SEQUENTIAL = 2,
+    MUTABLE_AGGREGATE = 3
+
 
 class Node(ABC):
     """
@@ -59,20 +93,40 @@ class Node(ABC):
     node name and the type of the node match.
     """
 
+    def __init__(self, node_name, node_input_type: NodeInputType, node_firing_type: NodeFiringType, node_state_type: NodeStateType):
+        self.__node_name__ = node_name
+        self.__node_input_type__ = node_input_type
+        self.__node_firing_type__ = node_firing_type
+        self.__node_state_type__ = node_state_type
+        self.__id__ = uuid.uuid4()
+
     def __str__(self):
         return self.__node_name__
 
+    def get_id(self):
+        return self.__id__
+
+    def get_node_input_type(self):
+        return self.__node_input_type__
+
+    def get_node_firing_type(self):
+        return self.__node_firing_type__
+
+    def get_node_state_type(self):
+        return self.__node_state_type__
+
     @abstractmethod
-    def get_and_flag(self):
-        raise NotImplementedError("Please implement this method")
+    def clone(self):
+        raise NotImplementedError("Please implement the clone method")
 
     def __hash__(self):
         """
         Hash code, defined as the hash code of the node name
 
         :return: Hash code
         """
-        return self.__node_name__.__hash__()
+
+        return self.__id__.__hash__()
 
     def __eq__(self, other):
         """
@@ -84,16 +138,16 @@ def __eq__(self, other):
         """
         return (
                 self.__class__ == other.__class__ and
+                self.__id__ == other.__id__ and
                 self.__node_name__ == other.__node_name__
         )
 
 
-class OrNode(Node):
+class EstimatorNode(Node):
     """
     Or node, which is the basic node that would be the equivalent of any SKlearn pipeline
     stage. This node is initialized with an estimator that needs to extend sklearn.BaseEstimator.
     """
-    __estimator__ = None
 
     def __init__(self, node_name: str, estimator: BaseEstimator):
         """
@@ -102,7 +156,8 @@ def __init__(self, node_name: str, estimator: BaseEstimator):
         :param node_name: Name of the node
         :param estimator: The base estimator
         """
-        self.__node_name__ = node_name
+
+        super().__init__(node_name, NodeInputType.OR, NodeFiringType.ANY, NodeStateType.IMMUTABLE)
         self.__estimator__ = estimator
 
     def get_estimator(self) -> BaseEstimator:
@@ -113,37 +168,33 @@ def get_estimator(self) -> BaseEstimator:
         """
         return self.__estimator__
 
-    def get_and_flag(self):
-        """
-        A flag to check if node is AND or not. By definition, this is NOT
-        an AND node.
-        :return: False, always
-        """
-        return False
+    def clone(self):
+        cloned_estimator = base.clone(self.__estimator__)
+        return EstimatorNode(self.__node_name__, cloned_estimator)
 
 
-class AndFunc(ABC):
-    """
-    Or nodes are init-ed from the
-    """
+class AndTransform(TransformerMixin, BaseEstimator):
+    @abstractmethod
+    def transform(self, xy_list: list) -> Xy:
+        raise NotImplementedError("Please implement this method")
 
+
+class GeneralTransform(TransformerMixin, BaseEstimator):
     @abstractmethod
-    def eval(self, xy_list: list) -> Xy:
+    def transform(self, xy: Xy) -> Xy:
         raise NotImplementedError("Please implement this method")
 
 
 class AndNode(Node):
-    __andfunc__ = None
-
-    def __init__(self, node_name: str, and_func: AndFunc):
-        self.__node_name__ = node_name
+    def __init__(self, node_name: str, and_func: AndTransform):
+        super().__init__(node_name, NodeInputType.AND, NodeFiringType.ANY, NodeStateType.STATELESS)
         self.__andfunc__ = and_func
 
-    def get_and_func(self) -> AndFunc:
+    def get_and_func(self) -> AndTransform:
         return self.__andfunc__
 
-    def get_and_flag(self):
-        return True
+    def clone(self):
+        return AndNode(self.__node_name__, self.__andfunc__)
 
 
 class Edge:

codeflare/pipelines/Runtime.py

@@ -1,11 +1,15 @@
 import ray
 
-from codeflare.pipelines.Datamodel import OrNode
+
+from codeflare.pipelines.Datamodel import EstimatorNode
 from codeflare.pipelines.Datamodel import AndNode
 from codeflare.pipelines.Datamodel import Edge
 from codeflare.pipelines.Datamodel import Pipeline
 from codeflare.pipelines.Datamodel import XYRef
 from codeflare.pipelines.Datamodel import Xy
+from codeflare.pipelines.Datamodel import NodeInputType
+from codeflare.pipelines.Datamodel import NodeStateType
+from codeflare.pipelines.Datamodel import NodeFiringType
 
 import sklearn.base as base
 from enum import Enum
@@ -18,47 +22,60 @@ class ExecutionType(Enum):
 
 
 @ray.remote
-def execute_or_node_inner(node: OrNode, train_mode: ExecutionType, Xy: XYRef):
+def execute_or_node_remote(node: EstimatorNode, train_mode: ExecutionType, xy_ref: XYRef):
     estimator = node.get_estimator()
     # Blocking operation -- not avoidable
-    X = ray.get(Xy.get_Xref())
-    y = ray.get(Xy.get_yref())
+    X = ray.get(xy_ref.get_Xref())
+    y = ray.get(xy_ref.get_yref())
 
+    # TODO: Can optimize the node pointers without replicating them
     if train_mode == ExecutionType.FIT:
+        cloned_node = node.clone()
+        prev_node_ptr = ray.put(node)
+
         if base.is_classifier(estimator) or base.is_regressor(estimator):
             # Always clone before fit, else fit is invalid
-            cloned_estimator = base.clone(estimator)
+            cloned_estimator = cloned_node.get_estimator()
             cloned_estimator.fit(X, y)
+
+            curr_node_ptr = ray.put(cloned_node)
             # TODO: For now, make yref passthrough - this has to be fixed more comprehensively
             res_Xref = ray.put(cloned_estimator.predict(X))
-            result = XYRef(res_Xref, Xy.get_yref())
+            result = XYRef(res_Xref, xy_ref.get_yref(), prev_node_ptr, curr_node_ptr, [xy_ref])
             return result
         else:
-            # No need to clone as it is a transform pass through on the fitted estimator
-            res_Xref = ray.put(estimator.fit_transform(X, y))
-            result = XYRef(res_Xref, Xy.get_yref())
+            cloned_estimator = cloned_node.get_estimator()
+            res_Xref = ray.put(cloned_estimator.fit_transform(X, y))
+            curr_node_ptr = ray.put(cloned_node)
+            result = XYRef(res_Xref, xy_ref.get_yref(), prev_node_ptr, curr_node_ptr, [xy_ref])
             return result
     elif train_mode == ExecutionType.SCORE:
+        cloned_node = node.clone()
+        prev_node_ptr = ray.put(node)
+
         if base.is_classifier(estimator) or base.is_regressor(estimator):
-            cloned_estimator = base.clone(estimator)
+            cloned_estimator = cloned_node.get_estimator()
             cloned_estimator.fit(X, y)
+            curr_node_ptr = ray.put(cloned_node)
             res_Xref = ray.put(cloned_estimator.score(X, y))
-            result = XYRef(res_Xref, Xy.get_yref())
+            result = XYRef(res_Xref, xy_ref.get_yref(), prev_node_ptr, curr_node_ptr, [xy_ref])
             return result
         else:
-            # No need to clone as it is a transform pass through on the fitted estimator
-            res_Xref = ray.put(estimator.fit_transform(X, y))
-            result = XYRef(res_Xref, Xy.get_yref())
+            cloned_estimator = cloned_node.get_estimator()
+            res_Xref = ray.put(cloned_estimator.fit_transform(X, y))
+            curr_node_ptr = ray.put(cloned_node)
+            result = XYRef(res_Xref, xy_ref.get_yref(), prev_node_ptr, curr_node_ptr, [xy_ref])
+            
             return result
     elif train_mode == ExecutionType.PREDICT:
         # Test mode does not clone as it is a simple predict or transform
         if base.is_classifier(estimator) or base.is_regressor(estimator):
             res_Xref = estimator.predict(X)
-            result = XYRef(res_Xref, Xy.get_yref())
+            result = XYRef(res_Xref, xy_ref.get_yref())
             return result
         else:
             res_Xref = estimator.transform(X)
-            result = XYRef(res_Xref, Xy.get_yref())
+            result = XYRef(res_Xref, xy_ref.get_yref())
             return result
 
 
@@ -68,7 +85,7 @@ def execute_or_node(node, pre_edges, edge_args, post_edges, mode: ExecutionType)
         exec_xyrefs = []
         for xy_ref_ptr in Xyref_ptrs:
             xy_ref = ray.get(xy_ref_ptr)
-            inner_result = execute_or_node_inner.remote(node, mode, xy_ref)
+            inner_result = execute_or_node_remote.remote(node, mode, xy_ref)
             exec_xyrefs.append(inner_result)
 
         for post_edge in post_edges:
@@ -78,29 +95,33 @@ def execute_or_node(node, pre_edges, edge_args, post_edges, mode: ExecutionType)
 
 
 @ray.remote
-def and_node_eval(and_func, Xyref_list):
+def execute_and_node_remote(node: AndNode, Xyref_list):
     xy_list = []
+    prev_node_ptr = ray.put(node)
     for Xyref in Xyref_list:
         X = ray.get(Xyref.get_Xref())
         y = ray.get(Xyref.get_yref())
         xy_list.append(Xy(X, y))
 
-    res_Xy = and_func.eval(xy_list)
+    cloned_node = node.clone()
+    curr_node_ptr = ray.put(cloned_node)
+
+    cloned_and_func = cloned_node.get_and_func()
+    res_Xy = cloned_and_func.transform(xy_list)
     res_Xref = ray.put(res_Xy.get_x())
     res_yref = ray.put(res_Xy.get_y())
-    return XYRef(res_Xref, res_yref)
+    return XYRef(res_Xref, res_yref, prev_node_ptr, curr_node_ptr, Xyref_list)
 
 
 def execute_and_node_inner(node: AndNode, Xyref_ptrs):
-    and_func = node.get_and_func()
     result = []
 
     Xyref_list = []
     for Xyref_ptr in Xyref_ptrs:
         Xyref = ray.get(Xyref_ptr)
         Xyref_list.append(Xyref)
 
-    Xyref_ptr = and_node_eval.remote(and_func, Xyref_list)
+    Xyref_ptr = execute_and_node_remote.remote(node, Xyref_list)
     result.append(Xyref_ptr)
     return result
 
@@ -136,9 +157,9 @@ def execute_pipeline(pipeline: Pipeline, mode: ExecutionType, in_args: dict):
         for node in nodes:
             pre_edges = pipeline.get_pre_edges(node)
             post_edges = pipeline.get_post_edges(node)
-            if not node.get_and_flag():
+            if node.get_node_input_type() == NodeInputType.OR:
                 execute_or_node(node, pre_edges, edge_args, post_edges, mode)
-            elif node.get_and_flag():
+            elif node.get_node_input_type() == NodeInputType.AND:
                 execute_and_node(node, pre_edges, edge_args, post_edges)
 
     out_args = {}

codeflare_pipelines.egg-info/SOURCES.txt

@@ -1,3 +1,4 @@
+README.md
 setup.py
 codeflare/__init__.py
 codeflare/pipelines/Datamodel.py