orm-profiler doesn't exists

[gachdavit]

Hello everybody,

I am trying to run Profiler for the Postgres. I want to collect metadata around the database and push the sample data. I decide to create my own DAG (I want to understand more about flow, etc) and execute it from Airflow.

import json
from datetime import timedelta

from airflow import DAG

try:
    from airflow.operators.python import PythonOperator
except ModuleNotFoundError:
    from airflow.operators.python_operator import PythonOperator

from airflow.utils.dates import days_ago

from metadata.ingestion.api.workflow import Workflow


default_args = {
    "owner": "user_name",
    "email": ["username@org.com"],
    "email_on_failure": False,
    "retries": 3,
    "retry_delay": timedelta(seconds=10),
    "execution_timeout": timedelta(minutes=60),
}

config: str = """
{
  "source": {
    "type": "postgres",
    "serviceName": "service_postgres",
    "serviceConnection": {
      "config": {
        "type": "Postgres",
        "username": "openmetadata_user",
        "password": "password",
        "hostPort": "postgresql:5432",
        "database": "postgres"
      }
    },
    "sourceConfig": {
      "config": {
        "type": "Profiler",
        "generateSampleData": true,
        "profileSample": 85
      }
    }
  },
  "processor": {
    "type": "orm-profiler",
    "config": {}
  },
  "sink": {
    "type": "metadata-rest",
    "config": {}
  },
  "workflowConfig": {
    "openMetadataServerConfig": {
      "hostPort": "http://openmetadata-server:8585/api",
      "authProvider": "openmetadata",
      "securityConfig": {
        "jwtToken": "<token>"
      }
    }
  }
}
"""


def metadata_ingestion_workflow():
    workflow_config = json.loads(config)
    workflow = Workflow.create(workflow_config)
    workflow.execute()
    workflow.raise_from_status()
    workflow.print_status()
    workflow.stop()


with DAG(
    "sample_usage",
    default_args=default_args,
    description="An example DAG which runs a OpenMetadata ingestion workflow",
    schedule_interval=timedelta(days=1),
    start_date=days_ago(1),
    is_paused_upon_creation=True,
    catchup=False,
) as dag:
    ingest_task = PythonOperator(
        task_id="ingest_using_recipe",
        python_callable=metadata_ingestion_workflow,
    )

Everything works fine, besides the processor. When I am providing orm-profiler, Python raises an exception:

Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.9/site-packages/metadata/utils/importer.py", line 99, in import_from_module
    obj = getattr(importlib.import_module(module_name), obj_name)
  File "/usr/local/lib/python3.9/importlib/__init__.py", line 127, in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
  File "<frozen importlib._bootstrap>", line 1030, in _gcd_import
  File "<frozen importlib._bootstrap>", line 1007, in _find_and_load
  File "<frozen importlib._bootstrap>", line 984, in _find_and_load_unlocked
ModuleNotFoundError: No module named 'metadata.ingestion.processor.orm_profiler'

I checked and orm_profiler.py does not exists at all in project, only query_parser.py is available here. Any ideas ?

Thanks !

[gachdavit]

I was using from metadata.ingestion.api.workflow import Workflow class, instead of from metadata.profiler.api.workflow import ProfilerWorkflow. I updated code. There is an example below and it works.

def metadata_ingestion_workflow():
    workflow_config = json.loads(config)
    # workflow = Workflow.create(workflow_config)
    workflow = ProfilerWorkflow.create(workflow_config)
    workflow.execute()
    workflow.raise_from_status()
    workflow.print_status()
    workflow.stop()