refactor out pipelineable inference operator

merlin/systems/dag/ops/faiss.py

@@ -28,10 +28,10 @@
 from merlin.core.protocols import Transformable
 from merlin.dag import ColumnSelector
 from merlin.schema import ColumnSchema, Schema
-from merlin.systems.dag.ops.operator import PipelineableInferenceOperator
+from merlin.systems.dag.ops.operator import InferenceOperator
 
 
-class QueryFaiss(PipelineableInferenceOperator):
+class QueryFaiss(InferenceOperator):
     """
     This operator creates an interface between a FAISS[1] Approximate Nearest Neighbors (ANN)
     Index and Triton Infrence Server. The operator allows users to perform different supported

merlin/systems/dag/ops/feast.py

@@ -1,4 +1,3 @@
-import json
 from typing import List
 
 import numpy as np
@@ -7,7 +6,7 @@
 from merlin.core.protocols import Transformable
 from merlin.dag import ColumnSelector
 from merlin.schema import ColumnSchema, Schema
-from merlin.systems.dag.ops.operator import PipelineableInferenceOperator
+from merlin.systems.dag.ops.operator import InferenceOperator
 
 # Feast_key: (numpy dtype, is_list, is_ragged)
 feast_2_numpy = {
@@ -20,7 +19,7 @@
 }
 
 
-class QueryFeast(PipelineableInferenceOperator):
+class QueryFeast(InferenceOperator):
     """
     The QueryFeast operator is responsible for ensuring that your feast feature store [1]
     can communicate correctly with tritonserver for the ensemble feast feature look ups.
@@ -191,75 +190,6 @@ def compute_input_schema(
         """Compute the input schema for the operator."""
         return self.input_schema
 
-    @classmethod
-    def from_config(cls, config, **kwargs) -> "QueryFeast":
-        """Create the operator from a config."""
-        parameters = json.loads(config.get("params", ""))
-        entity_id = parameters["entity_id"]
-        entity_view = parameters["entity_view"]
-        entity_column = parameters["entity_column"]
-        repo_path = parameters["feast_repo_path"]
-        features = parameters["features"]
-        mh_features = parameters["mh_features"]
-        in_dict = json.loads(config.get("input_dict", "{}"))
-        out_dict = json.loads(config.get("output_dict", "{}"))
-        include_id = parameters["include_id"]
-        output_prefix = parameters["output_prefix"]
-
-        in_schema = Schema([])
-        for col_name, col_rep in in_dict.items():
-            in_schema[col_name] = ColumnSchema(
-                col_name,
-                dtype=col_rep["dtype"],
-                is_list=col_rep["is_list"],
-                is_ragged=col_rep["is_ragged"],
-            )
-        out_schema = Schema([])
-        for col_name, col_rep in out_dict.items():
-            out_schema[col_name] = ColumnSchema(
-                col_name,
-                dtype=col_rep["dtype"],
-                is_list=col_rep["is_list"],
-                is_ragged=col_rep["is_ragged"],
-            )
-
-        return QueryFeast(
-            repo_path,
-            entity_id,
-            entity_view,
-            entity_column,
-            features,
-            mh_features,
-            in_schema,
-            out_schema,
-            include_id,
-            output_prefix,
-        )
-
-    def export(
-        self,
-        path: str,
-        input_schema: Schema,
-        output_schema: Schema,
-        params: dict = None,
-        node_id: int = None,
-        version: int = 1,
-        backend: str = "ensemble",
-    ):
-        params = params or {}
-        self_params = {
-            "entity_id": self.entity_id,
-            "entity_view": self.entity_view,
-            "entity_column": self.entity_column,
-            "features": self.features,
-            "mh_features": self.mh_features,
-            "feast_repo_path": self.repo_path,
-            "include_id": self.include_id,
-            "output_prefix": self.output_prefix,
-        }
-        self_params.update(params)
-        return super().export(path, input_schema, output_schema, self_params, node_id, version)
-
     def transform(
         self, col_selector: ColumnSelector, transformable: Transformable
     ) -> Transformable:

merlin/systems/dag/ops/fil.py

@@ -33,10 +33,10 @@
     treelite_sklearn,
     xgboost,
 )
-from merlin.systems.dag.ops.operator import InferenceOperator, PipelineableInferenceOperator
+from merlin.systems.dag.ops.operator import InferenceOperator
 
 
-class PredictForest(PipelineableInferenceOperator):
+class PredictForest(InferenceOperator):
     """Operator for running inference on Forest models.
 
     This works for gradient-boosted decision trees (GBDTs) and Random forests (RF).

merlin/systems/dag/ops/implicit.py

@@ -14,15 +14,14 @@
 # limitations under the License.
 #
 import importlib
-import json
 import pathlib
 
 import numpy as np
 
 from merlin.core.protocols import Transformable
 from merlin.dag import ColumnSelector
 from merlin.schema import ColumnSchema, Schema
-from merlin.systems.dag.ops.operator import PipelineableInferenceOperator
+from merlin.systems.dag.ops.operator import InferenceOperator
 
 try:
     import implicit
@@ -36,7 +35,7 @@
     implicit = None
 
 
-class PredictImplicit(PipelineableInferenceOperator):
+class PredictImplicit(InferenceOperator):
     """Operator for running inference on Implicit models.."""
 
     def __init__(self, model, num_to_recommend: int = 10, **kwargs):
@@ -90,73 +89,6 @@ def compute_output_schema(
     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."""
-        node_name = f"{node_id}_{self.export_name}" if node_id is not None else self.export_name
-
-        if backend == "ensemble":
-            artifact_path = pathlib.Path(path) / node_name / str(version)
-        else:
-            artifact_path = pathlib.Path(path) / "executor_model" / str(version) / "ensemble"
-
-        artifact_path.mkdir(parents=True, exist_ok=True)
-        model_path = artifact_path / "model.npz"
-        self.model.save(str(model_path))
-
-        if backend == "ensemble":
-            params = params or {}
-            params["model_module_name"] = self.model.__module__
-            params["model_class_name"] = self.model.__class__.__name__
-            params["num_to_recommend"] = self.num_to_recommend
-            return super().export(
-                path,
-                input_schema,
-                output_schema,
-                params=params,
-                node_id=node_id,
-                version=version,
-            )
-        else:
-            return ({}, [])
-
-    @classmethod
-    def from_config(cls, config: dict, **kwargs) -> "PredictImplicit":
-        """Instantiate the class from a dictionary representation.
-
-        Expected config structure:
-        {
-            "input_dict": str  # JSON dict with input names and schemas
-            "params": str  # JSON dict with params saved at export
-        }
-
-        """
-        params = json.loads(config["params"])
-
-        model_repository = kwargs["model_repository"]
-        model_name = kwargs["model_name"]
-        model_version = kwargs["model_version"]
-
-        # load implicit model
-        model_module_name = params["model_module_name"]
-        model_class_name = params["model_class_name"]
-        model_module = importlib.import_module(model_module_name)
-        model_cls = getattr(model_module, model_class_name)
-        model_file = pathlib.Path(model_repository) / model_name / str(model_version) / "model.npz"
-        model = model_cls.load(str(model_file))
-
-        num_to_recommend = params["num_to_recommend"]
-
-        return cls(model, num_to_recommend=num_to_recommend)
-
     def transform(
         self, col_selector: ColumnSelector, transformable: Transformable
     ) -> Transformable: