postgresql_to_datahub_rest.py

# Purpose: Simple programmatic DataHub pipline example
# Author: Gary A. Stafford
# Date: March 2022
# Reference: https://github.com/datahub-project/datahub/blob/master/metadata-ingestion/examples/library/programatic_pipeline.py

from datahub.ingestion.run.pipeline import Pipeline

# The pipeline configuration is similar to the recipe YAML files provided to the CLI tool.
pipeline = Pipeline.create(
    {
        "run_id": "postgres-run",
        "source": {
            "type": "postgres",
            "config": {
                "host_port": "demo-instance.abcd1234.us-east-1.rds.amazonaws.com:5432",
                "database": "tickit",
                "username": "datahub",
                "password": "My5up3r53cr3tPa55w0rd",
                "env": "DEV",
                "profiling": {
                    "enabled": "true"
                }
            }
        },
        "transformers": [
            {
                "type": "simple_add_dataset_tags",
                "config": {
                    "tag_urns": [
                        f"urn:li:tag:AWS",
                        f"urn:li:tag:111222333444",
                        f"urn:li:tag:us-east-1"
                    ]
                }
            },
            {
                "type": "pattern_add_dataset_terms",
                "config": {
                    "term_pattern": {
                        "rules": {
                            ".*users.*": [
                                "urn:li:glossaryTerm:Classification.Sensitive"
                            ]
                        }
                    }
                }
            },
            {
                "type": "simple_add_dataset_ownership",
                "config": {
                    "owner_urns": [
                        f"urn:li:corpuser:Database Administrators"
                    ],
                    "ownership_type": "DATAOWNER"
                }
            }
        ],
        "sink": {
            "type": "datahub-rest",
            "config": {
                "server": "http://192.168.111.222:33333"
            }
        }
    }
)

# Run the pipeline and report the results.
pipeline.run()
pipeline.pretty_print_summary()