OpenMetadata Ingestion Framework

Warnings

• breaking A version mismatch between the `openmetadata-ingestion` Python package and the OpenMetadata server can lead to `ClientInitializationError`, `ValidationError`, or other unexpected behavior. The ingestion package version (e.g., `1.12.5.1`) must match the server version (e.g., `~=1.12.5`).
  Fix: Ensure that `openmetadata-ingestion` is installed with a version compatible with your OpenMetadata server version, typically using a tilde-equals `~=` version specifier (e.g., `pip install "openmetadata-ingestion~=1.12.5"`).
• breaking Schema changes introduced in OpenMetadata minor releases (e.g., adding `queryStatementSource`, `queryParserConfig`, `statusLookbackDays` fields) require upgrading the ingestion framework. Running an older ingestion framework with a newer server can cause `ValidationError` during pipeline deployment or Airflow DAG build failures.
  Fix: Always upgrade the `openmetadata-ingestion` package to match your OpenMetadata server version, especially after server upgrades, to ensure schema compatibility.
• breaking The OpenLineage Kinesis connection schema was refactored in version 1.11.12. Existing configurations with flat Kafka-specific fields (e.g., `brokersUrl`, `topicName`) must be migrated to a nested `brokerConfig` object.
  Fix: Update your OpenLineage Kinesis connector configurations to use the new `brokerConfig` object structure.
• breaking The REST connector configuration changed in version 1.11.11 to support OpenAPI schemas from local files (JSON/YAML) in addition to HTTP URLs via the `openAPISchemaConnection` union type. Existing configurations might require updates.
  Fix: Refer to the official documentation for the updated REST connector connection structure and adjust your YAML configurations accordingly.
• gotcha The `openmetadata-ingestion` library requires Python versions 3.9, 3.10, or 3.11. Using unsupported Python versions might lead to compatibility issues or errors.
  Fix: Ensure your Python environment is running a supported version (3.9, 3.10, or 3.11). It's recommended to use a virtual environment.

Install

• pip install "openmetadata-ingestion==1.12.5.1" Basic Installation
• pip install "openmetadata-ingestion[all]==1.12.5.1" Installation with All Connectors
• pip install "openmetadata-ingestion[postgres,snowflake]==1.12.5.1" Installation with specific database connectors (e.g., PostgreSQL, Snowflake)

Imports

• MetadataWorkflow

  from metadata.workflow.metadata import MetadataWorkflow

• OpenMetadata

  from metadata.ingestion.ometa.ometa_api import OpenMetadata

Quickstart

This quickstart demonstrates a basic metadata ingestion workflow using the `MetadataWorkflow` class. It shows how to define a minimal workflow configuration in YAML, including the OpenMetadata server connection details and using an environment variable for the JWT token for authentication. Replace `sample-data` with your actual connector configuration (e.g., `mysql`, `snowflake`). Ensure your OpenMetadata server is running and accessible.

import os
import yaml
from metadata.workflow.metadata import MetadataWorkflow

# Define your OpenMetadata server connection and authentication
# It's recommended to use environment variables for sensitive data like JWT tokens
openmetadata_server_config = {
    "hostPort": "http://localhost:8585/api",
    "authProvider": "openmetadata",
    "securityConfig": {
        "jwtToken": os.environ.get('OPENMETADATA_JWT_TOKEN', 'YOUR_OM_JWT_TOKEN_HERE')
    }
}

# Example: A minimal YAML configuration for ingesting metadata from a dummy source
# In a real scenario, this would be a full connector config (e.g., MySQL, Snowflake)
workflow_config_yaml = f"""
source:
  type: "sample-data"
  serviceName: "sample_metadata"
sink:
  type: "metadata-rest"
  config: {}
workflowConfig:
  openMetadataServerConfig:
    hostPort: {openmetadata_server_config['hostPort']}
    authProvider: {openmetadata_server_config['authProvider']}
    securityConfig:
      jwtToken: {openmetadata_server_config['securityConfig']['jwtToken']}
"""

# Load the YAML configuration
workflow_config = yaml.safe_load(workflow_config_yaml)

# Create and execute the workflow
print("Starting Metadata Ingestion Workflow...")
workflow = MetadataWorkflow.create(workflow_config)
try:
    workflow.execute()
    print("Metadata Ingestion Workflow completed successfully.")
except Exception as err:
    print(f"Error during metadata ingestion: {err}")
finally:
    workflow.print_status()
    workflow.stop()