Adding lineage first commit, needs a documentation/ADR

Author: raghukiran1224

codeflare/pipelines/Datamodel.py

@@ -1,6 +1,9 @@
 from sklearn.base import BaseEstimator
 from abc import ABC, abstractmethod
 
+import sklearn.base as base
+import uuid
+
 
 class Xy:
     """
@@ -35,9 +38,11 @@ class XYRef:
     computed), these holders are essential to the pipeline constructs.
     """
 
-    def __init__(self, Xref, yref):
+    def __init__(self, Xref, yref, node=None, prev_Xyrefs = None):
         self.__Xref__ = Xref
         self.__yref__ = yref
+        self.__noderef__ = node
+        self.__prev_Xyrefs__ = prev_Xyrefs
 
     def get_Xref(self):
         """
@@ -51,6 +56,12 @@ def get_yref(self):
         """
         return self.__yref__
 
+    def get_noderef(self):
+        return self.__noderef__
+
+    def get_prev_xyrefs(self):
+        return self.__prev_Xyrefs__
+
 
 class Node(ABC):
     """
@@ -62,17 +73,24 @@ class Node(ABC):
     def __str__(self):
         return self.__node_name__
 
+    def get_id(self):
+        return self.__id__
+
     @abstractmethod
     def get_and_flag(self):
         raise NotImplementedError("Please implement this method")
 
+    @abstractmethod
+    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,6 +102,7 @@ def __eq__(self, other):
         """
         return (
                 self.__class__ == other.__class__ and
+                self.__id__ == other.__id__ and
                 self.__node_name__ == other.__node_name__
         )
 
@@ -93,7 +112,6 @@ class OrNode(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):
         """
@@ -104,6 +122,7 @@ def __init__(self, node_name: str, estimator: BaseEstimator):
         """
         self.__node_name__ = node_name
         self.__estimator__ = estimator
+        self.__id__ = uuid.uuid4()
 
     def get_estimator(self) -> BaseEstimator:
         """
@@ -121,6 +140,10 @@ def get_and_flag(self):
         """
         return False
 
+    def clone(self):
+        cloned_estimator = base.clone(self.__estimator__)
+        return OrNode(self.__node_name__, cloned_estimator)
+
 
 class AndFunc(ABC):
     """
@@ -133,18 +156,20 @@ def eval(self, xy_list: list) -> Xy:
 
 
 class AndNode(Node):
-    __andfunc__ = None
-
     def __init__(self, node_name: str, and_func: AndFunc):
         self.__node_name__ = node_name
         self.__andfunc__ = and_func
+        self.__id__ = uuid.uuid4()
 
     def get_and_func(self) -> AndFunc:
         return self.__andfunc__
 
     def get_and_flag(self):
         return True
 
+    def clone(self):
+        return AndNode(self.__node_name__, self.__andfunc__)
+
 
 class Edge:
     __from_node__ = None

codeflare/pipelines/Runtime.py

@@ -18,47 +18,53 @@ class ExecutionType(Enum):
 
 
 @ray.remote
-def execute_or_node_inner(node: OrNode, train_mode: ExecutionType, Xy: XYRef):
+def execute_or_node_inner(node: OrNode, 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())
 
     if train_mode == ExecutionType.FIT:
+        cloned_node = node.clone()
+        node_ptr = ray.put(cloned_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)
             # 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(), 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))
+            result = XYRef(res_Xref, xy_ref.get_yref(), node_ptr, [xy_ref])
             return result
     elif train_mode == ExecutionType.SCORE:
+        cloned_node = node.clone()
+        node_ptr = ray.put(cloned_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)
             res_Xref = ray.put(cloned_estimator.score(X, y))
-            result = XYRef(res_Xref, Xy.get_yref())
+            result = XYRef(res_Xref, xy_ref.get_yref(), 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))
+            result = XYRef(res_Xref, xy_ref.get_yref(), 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
 
 
@@ -78,29 +84,32 @@ def execute_or_node(node, pre_edges, edge_args, post_edges, mode: ExecutionType)
 
 
 @ray.remote
-def and_node_eval(and_func, Xyref_list):
+def and_node_eval(node: AndNode, Xyref_list):
     xy_list = []
     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()
+    node_ptr = ray.put(cloned_node)
+
+    cloned_and_func = cloned_node.get_and_func()
+    res_Xy = cloned_and_func.eval(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, 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 = and_node_eval.remote(node, Xyref_list)
     result.append(Xyref_ptr)
     return result
 

codeflare_pipelines.egg-info/SOURCES.txt

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