Fix #2894 - Profiler Processor & Metrics

ingestion/src/metadata/config/workflow.py

@@ -17,6 +17,7 @@
 from typing import Type, TypeVar
 
 from metadata.ingestion.api.common import DynamicTypedConfig, WorkflowContext
+from metadata.ingestion.api.processor import Processor
 from metadata.ingestion.api.sink import Sink
 from metadata.ingestion.api.source import Source
 from metadata.ingestion.ometa.openmetadata_rest import MetadataServerConfig
@@ -89,7 +90,7 @@ def get_sink(
     _from: str = "ingestion",
 ) -> Sink:
     """
-    Helps us fetching and importing the sink class.
+    Helps us to fetch and importing the sink class.
 
     By default, we will pick it up from `ingestion`.
 
@@ -114,3 +115,43 @@ def get_sink(
     logger.debug(f"Sink type: {sink_type}, {sink_class} configured")
 
     return sink
+
+
+def get_processor(
+    processor_type: str,
+    context: WorkflowContext,
+    processor_config: DynamicTypedConfig,
+    metadata_config: MetadataServerConfig,
+    _from: str = "ingestion",
+    **kwargs,
+) -> Processor:
+    """
+    Helps us to fetch and import the Processor class.
+
+    By default, we will pick it up from `ingestion`
+
+    We allow to pass any other specific object we may require.
+    E.g., for the ORM Profiler we need a Session to reach
+    the source tables.
+
+    :param processor_type: Type specified in the config, e.g., metadata-rest
+    :param context: Workflow related information
+    :param processor_config: Specific Processor configurations, such as the profiler and tests
+    :param metadata_config: Metadata server configurations
+    :param _from: From where do we load the sink class. Ingestion by default.
+    """
+    processor_class = get_class(
+        "metadata.{}.processor.{}.{}Processor".format(
+            _from,
+            fetch_type_class(processor_type, is_file=True),
+            fetch_type_class(processor_type, is_file=False),
+        )
+    )
+
+    processor: Processor = processor_class.create(
+        processor_config.dict().get("config", {}), metadata_config, context, **kwargs
+    )
+
+    logger.debug(f"Sink type: {processor_type}, {processor_class} configured")
+
+    return processor

ingestion/src/metadata/ingestion/api/processor.py

@@ -41,7 +41,11 @@ class Processor(Closeable, Generic[Entity], metaclass=ABCMeta):
     @classmethod
     @abstractmethod
     def create(
-        cls, config_dict: dict, metadata_config_dict: dict, ctx: WorkflowContext
+        cls,
+        config_dict: dict,
+        metadata_config_dict: dict,
+        ctx: WorkflowContext,
+        **kwargs
     ) -> "Processor":
         pass
 

ingestion/src/metadata/ingestion/processor/query_parser.py

@@ -67,7 +67,11 @@ def __init__(
 
     @classmethod
     def create(
-        cls, config_dict: dict, metadata_config_dict: dict, ctx: WorkflowContext
+        cls,
+        config_dict: dict,
+        metadata_config_dict: dict,
+        ctx: WorkflowContext,
+        **kwargs
     ):
         config = QueryParserProcessorConfig.parse_obj(config_dict)
         metadata_config = MetadataServerConfig.parse_obj(metadata_config_dict)

ingestion/src/metadata/orm_profiler/api/models.py

@@ -15,12 +15,14 @@
 We need to define this class as we end up having
 multiple profilers per table and columns.
 """
-from typing import List
+from typing import List, Optional
 
 from pydantic import BaseModel
 
 from metadata.generated.schema.entity.data.table import Table
 from metadata.orm_profiler.profiles.core import Profiler
+from metadata.orm_profiler.profiles.models import ProfilerDef
+from metadata.orm_profiler.validations.models import TestDef
 
 
 class WorkflowResult(BaseModel):
@@ -37,3 +39,25 @@ class ProfilerResult(WorkflowResult):
     table: Table  # Table Entity
     table_profiler: Profiler  # Profiler with table results
     column_profilers: List[ColumnProfiler]  # Profiler with col results
+
+
+class ProfilerProcessorConfig(BaseModel):
+    """
+    Defines how we read the processor information
+    from the workflow JSON definition
+    """
+
+    profiler: Optional[ProfilerDef] = None
+    tests: Optional[TestDef] = None
+
+
+class ProfileAndTests(BaseModel):
+    """
+    ORM Profiler processor response.
+
+    For a given table, return all profilers and
+    the ran tests, if any.
+    """
+
+    profile: ProfilerResult
+    tests: Optional[TestDef] = None