'DatabaseServiceProfilerPipeline' object has no attribute 'dbtConfigSource'

[pmbrull]

When trying to run the Profiler workflow I am getting the following error when running via the CLI:

[2022-07-07, 14:59:32 UTC] {taskinstance.py:1774} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/operators/python.py", line 174, in execute
    return_value = self.execute_callable()
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/operators/python.py", line 188, in execute_callable
    return self.python_callable(*self.op_args, **self.op_kwargs)
  File "/opt/airflow/dags/repo/dags/dev/openmetadata_snowflake_profiler.py", line 70, in profiler_workflow
    workflow = Workflow.create(workflow_config)
  File "/home/airflow/.local/lib/python3.7/site-packages/metadata/ingestion/api/workflow.py", line 158, in create
    return cls(config)
  File "/home/airflow/.local/lib/python3.7/site-packages/metadata/ingestion/api/workflow.py", line 79, in __init__
    self.config.source.dict(), metadata_config
  File "/home/airflow/.local/lib/python3.7/site-packages/metadata/ingestion/source/database/snowflake.py", line 105, in create
    return cls(config, metadata_config)
  File "/home/airflow/.local/lib/python3.7/site-packages/metadata/ingestion/source/database/snowflake.py", line 95, in __init__
    super().__init__(config, metadata_config)
  File "/home/airflow/.local/lib/python3.7/site-packages/metadata/ingestion/source/database/common_db_source.py", line 95, in __init__
    super().__init__()
  File "/home/airflow/.local/lib/python3.7/site-packages/metadata/ingestion/source/database/database_service.py", line 192, in __init__
    dbt_details = get_dbt_details(self.source_config.dbtConfigSource)
AttributeError: 'DatabaseServiceProfilerPipeline' object has no attribute 'dbtConfigSource'

My workflow looks good as per the docs:

source:
  type: snowflake
  serviceName: snowflake
  serviceConnection:
    config:
      type: Snowflake
      username: ...
      password: ...
      warehouse: ...
      account: ...
  sourceConfig:
    config:
      type: Profiler
      fqnFilterPattern:
        includes:
        - snowflake.SHOP_DB.DBT_JAFFLE.*
      generateSampleData: true
processor:
  type: orm-profiler
  config:
    test_suite:
        name: demo_data
        tests:
          - table: snowflake.SHOP_DB.DBT_JAFFLE.CUSTOMERS
            table_tests:
              - testCase:
                  config:
                    value: 80
                  tableTestType: tableRowCountToEqual
            column_tests:
              - columnName: CUSTOMER_ID
                testCase:
                  config:
                    minValue: 1
                    maxValue: 80
                  columnTestType: columnValuesToBeBetween
sink:
  type: metadata-rest
  config: {}
workflowConfig:
  openMetadataServerConfig:
    hostPort: http://localhost:8585/api
    authProvider: no-auth

What am I missing?

[pmbrull]

While the workflow YAML is correct, we need to take into account which CLI command are we using. While the metadata ingestion workflow is executed as metadata ingest, the Profiler workflow needs to be run as metadata profile.

Link to the docs.

---

Running this via an Airflow DAG (or with any approach related to importing the Workflow class from Python), the difference is in the type of Workflow class being used.

• To ingest metadata, we need to import from metadata.ingestion.api.workflow import Workflow
• For the Profiler, the import is from metadata.orm_profiler.api.workflow import ProfilerWorkflow