Remove ensemble specific code

merlin/systems/dag/__init__.py

@@ -18,4 +18,3 @@
 # flake8: noqa
 from .ensemble import Ensemble
 from .node import Node
-from .op_runner import OperatorRunner

merlin/systems/dag/node.py

@@ -23,31 +23,11 @@
 class InferenceNode(Node):
     """Specialized node class used in Triton Ensemble DAGs"""
 
-    def exportable(self, backend: str = None):
-        """
-        Determine whether the current node's operator is exportable for a given back-end
-
-        Parameters
-        ----------
-        backend : str, optional
-            The Merlin Systems (not Triton) back-end to use,
-            either "ensemble" or "executor", by default None
-
-        Returns
-        -------
-        bool
-            True if the node's operator is exportable for the supplied back-end
-        """
-        backends = getattr(self.op, "exportable_backends", [])
-
-        return hasattr(self.op, "export") and backend in backends
-
     def export(
         self,
         output_path: Union[str, os.PathLike],
         node_id: int = None,
         version: int = 1,
-        backend="ensemble",
     ):
         """
         Export a Triton config directory for this node.
@@ -72,7 +52,6 @@ def export(
             self.output_schema,
             node_id=node_id,
             version=version,
-            backend=backend,
         )
 
     @property

merlin/systems/dag/ops/faiss.py

@@ -13,13 +13,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-
-import json
 import os
-import pathlib
 from pathlib import Path
 from shutil import copy2
-from typing import Dict, List, Tuple
 
 import faiss
 import numpy as np
@@ -61,7 +57,7 @@ def __init__(self, index_path, topk=10):
         self.topk = topk
         self._index = None
 
-    def load_artifacts(self, artifact_path):
+    def load_artifacts(self, artifact_path: str) -> None:
         filename = Path(self.index_path).name
         path_artifact = Path(artifact_path)
         if path_artifact.is_file():
@@ -74,93 +70,20 @@ def load_artifacts(self, artifact_path):
             index = faiss.index_cpu_to_gpu(res, 0, index)
         self._index = index
 
-    @classmethod
-    def from_config(cls, config: dict, **kwargs) -> "QueryFaiss":
-        """
-        Instantiate a class object given a config.
-
-        Parameters
-        ----------
-        config : dict
-
-
-        Returns
-        -------
-        QueryFaiss
-            class object instantiated with config values
-        """
-        parameters = json.loads(config.get("params", ""))
-        index_path = parameters["index_path"]
-        topk = parameters["topk"]
-
-        operator = QueryFaiss(index_path, topk=topk)
-        operator.load_artifacts(index_path)
-
-        return operator
-
-    @property
-    def exportable_backends(self):
-        return ["ensemble", "executor"]
-
-    def export(
-        self,
-        path: str,
-        input_schema: Schema,
-        output_schema: Schema,
-        params: dict = None,
-        node_id: int = None,
-        version: int = 1,
-        backend: str = "ensemble",
-    ) -> Tuple[Dict, List]:
-        """
-        Export the class object as a config and all related files to the user defined path.
+    def save_artifacts(self, artifact_path: str) -> None:
+        index_filename = os.path.basename(os.path.realpath(self.index_path))
+        new_index_path = Path(artifact_path) / index_filename
+        copy2(self.index_path, new_index_path)
 
-        Parameters
-        ----------
-        path : str
-            Artifact export path
-        input_schema : Schema
-            A schema with information about the inputs to this operator
-        output_schema : Schema
-            A schema with information about the outputs of this operator
-        params : dict, optional
-            Parameters dictionary of key, value pairs stored in exported config, by default None
-        node_id : int, optional
-            The placement of the node in the graph (starts at 1), by default None
-        version : int, optional
-            The version of the model, by default 1
+    def __getstate__(self) -> dict:
+        """Return state of instance when pickled.
 
         Returns
         -------
-        Ensemble_config: dict
-        Node_configs: list
+        dict
+            Returns object state excluding index attribute.
         """
-        params = params or {}
-
-        self_params = {
-            # TODO: Write the (relative) path from inside the export directory
-            "index_path": self.index_path,
-            "topk": self.topk,
-        }
-        self_params.update(params)
-        index_filename = os.path.basename(os.path.realpath(self.index_path))
-
-        if backend == "ensemble":
-            full_path = (
-                pathlib.Path(path) / f"{node_id}_{QueryFaiss.__name__.lower()}" / str(version)
-            )
-        else:
-            full_path = pathlib.Path(path) / "executor_model" / str(version) / "ensemble"
-
-        new_index_path = full_path / index_filename
-        full_path.mkdir(parents=True, exist_ok=True)
-        copy2(self.index_path, new_index_path)
-        self.index_path = str(new_index_path)
-
-        if backend == "ensemble":
-            return super().export(path, input_schema, output_schema, self_params, node_id, version)
-        else:
-            return ({}, [])
+        return {k: v for k, v in self.__dict__.items() if k != "_index"}
 
     def transform(
         self, col_selector: ColumnSelector, transformable: Transformable

merlin/systems/dag/ops/fil.py

@@ -88,39 +88,6 @@ def compute_input_schema(
         """Return the input schema representing the input columns this operator expects to use."""
         return self.input_schema
 
-    @property
-    def exportable_backends(self):
-        return ["ensemble", "executor"]
-
-    def export(
-        self,
-        path: str,
-        input_schema: Schema,
-        output_schema: Schema,
-        params: dict = None,
-        node_id: int = None,
-        version: int = 1,
-        backend: str = "ensemble",
-    ):
-        """Export the class and related files to the path specified."""
-        fil_model_config = self.fil_op.export(
-            path,
-            input_schema,
-            output_schema,
-            params=params,
-            node_id=node_id,
-            version=version,
-        )
-
-        return fil_model_config
-
-    @property
-    def fil_model_name(self):
-        return self._fil_model_name
-
-    def set_fil_model_name(self, fil_model_name):
-        self._fil_model_name = fil_model_name
-
     def transform(
         self, col_selector: ColumnSelector, transformable: Transformable
     ) -> Transformable:
@@ -147,10 +114,6 @@ def transform(
 
         return type(transformable)(outputs)
 
-    def load_artifacts(self, artifact_path):
-        # need variable that tells me what type of model this is.
-        self.fil_op.load_model(artifact_path)
-
 
 class FIL(InferenceOperator):
     """Operator for Forest Inference Library (FIL) models.
@@ -265,25 +228,6 @@ def compute_output_schema(
         """Returns output schema for FIL op"""
         return Schema([ColumnSchema("output__0", dtype=np.float32)])
 
-    def export(
-        self,
-        path,
-        input_schema,
-        output_schema,
-        params: dict = None,
-        node_id=None,
-        version=1,
-    ):
-        """Export the model to the supplied path. Returns the config"""
-        node_name = f"{node_id}_{self.export_name}" if node_id is not None else self.export_name
-        node_export_path = pathlib.Path(path) / node_name
-        version_path = node_export_path / str(version)
-        version_path.mkdir(parents=True, exist_ok=True)
-
-        self.fil_model_class.save(version_path)
-
-        return version_path
-
     def load_model(self, version_path):
         version_path = pathlib.Path(version_path)
         self.fil_model_class.load(version_path)
@@ -305,6 +249,12 @@ def save(self, version_path):
         Save model to version_path
         """
 
+    @abstractmethod
+    def load(self, version_path):
+        """
+        Load model from path
+        """
+
     @property
     @abstractmethod
     def num_classes(self):

merlin/systems/dag/ops/implicit.py

@@ -66,6 +66,10 @@ def load_artifacts(self, artifact_path: str):
 
         self.model = model_cls.load(str(model_file))
 
+    def save_artifacts(self, artifact_path: str):
+        model_path = pathlib.Path(artifact_path) / "model.npz"
+        self.model.save(str(model_path))
+
     def compute_input_schema(
         self,
         root_schema: Schema,
@@ -85,10 +89,6 @@ def compute_output_schema(
         """Return the output schema representing the columns this operator returns."""
         return Schema([ColumnSchema("ids", dtype="int64"), ColumnSchema("scores", dtype="float64")])
 
-    @property
-    def exportable_backends(self):
-        return ["ensemble", "executor"]
-
     def transform(
         self, col_selector: ColumnSelector, transformable: Transformable
     ) -> Transformable: