Platform Integrations

metapod extracts metadata locally. You choose where it goes.

Dataflix (built-in)

metapod push --target https://dataflix.macalab.org \
             --project my-project \
             --api-key $API_KEY

Dataflix converts probe JSON to DDL, builds a lineage graph, and provides a visual explorer with 5 Knowledge Graph views.

DataHub

Convert metapod output to DataHub Metadata Change Events (MCEs):

"""Push metapod probe results to DataHub."""
import json
from pathlib import Path
from datahub.emitter.rest_emitter import DatahubRestEmitter
from datahub.metadata.schema_classes import (
    DatasetSnapshotClass, DatasetPropertiesClass, SchemaMetadataClass,
    SchemaFieldClass, MySqlDDLClass,
)

emitter = DatahubRestEmitter("http://datahub-gms:8080")

# Push tables
tables = json.load(open("output/oracle/metadata/tables.json"))
for t in tables["records"]:
    platform = tables["source"]["type"]
    urn = f"urn:li:dataset:(urn:li:dataPlatform:{platform},{t['table_name']},PROD)"
    snapshot = DatasetSnapshotClass(
        urn=urn,
        aspects=[
            DatasetPropertiesClass(
                name=t["table_name"],
                customProperties={"owner": t.get("owner", ""), "row_count": str(t.get("row_count", ""))},
            ),
        ],
    )
    emitter.emit_mce(snapshot)

# Push columns
columns = json.load(open("output/oracle/metadata/columns.json"))
col_map = {}
for c in columns["records"]:
    col_map.setdefault(c["table_name"], []).append(c)

for table_name, cols in col_map.items():
    urn = f"urn:li:dataset:(urn:li:dataPlatform:{platform},{table_name},PROD)"
    fields = [
        SchemaFieldClass(
            fieldPath=c["column_name"],
            nativeDataType=c.get("data_type", ""),
            type={"type": {"com.linkedin.schema.StringType": {}}},
        )
        for c in sorted(cols, key=lambda x: x.get("ordinal_position", 0))
    ]
    emitter.emit_mce(DatasetSnapshotClass(
        urn=urn,
        aspects=[SchemaMetadataClass(
            schemaName=table_name, platform=f"urn:li:dataPlatform:{platform}",
            hash="", platformSchema=MySqlDDLClass(tableSchema=""),
            fields=fields,
        )],
    ))

OpenMetadata

Use the OpenMetadata REST API to push probe results:

"""Push metapod probe results to OpenMetadata."""
import json
import httpx

OM_URL = "http://openmetadata:8585/api/v1"
TOKEN = "your-jwt-token"
headers = {"Authorization": f"Bearer {TOKEN}"}

tables = json.load(open("output/postgres/metadata/tables.json"))
for t in tables["records"]:
    httpx.put(
        f"{OM_URL}/tables",
        headers=headers,
        json={
            "name": t["table_name"],
            "databaseSchema": f"postgres.public",
            "columns": [],  # populate from columns.json
            "tableType": "Regular",
        },
    )

Apache Atlas / Purview

metapod already has a Purview probe (purview.metadata.*). For bidirectional sync, push results back:

"""Push metapod results to Purview as custom entities."""
import json
import httpx

PURVIEW_ACCOUNT = "my-purview"
token = get_purview_token()  # OAuth2 client_credentials flow

tables = json.load(open("output/oracle/metadata/tables.json"))
for t in tables["records"]:
    httpx.post(
        f"https://{PURVIEW_ACCOUNT}.purview.azure.com/catalog/api/atlas/v2/entity",
        headers={"Authorization": f"Bearer {token}"},
        json={
            "entity": {
                "typeName": "azure_sql_table",
                "attributes": {
                    "qualifiedName": f"oracle://{t['table_name']}",
                    "name": t["table_name"],
                },
            }
        },
    )

Custom REST API

Generic pattern for pushing to any REST endpoint:

"""Push all probe results to a custom API."""
import json
import httpx
from pathlib import Path

API_URL = "https://my-catalog.example.com/api/ingest"
API_KEY = "your-key"

for jf in Path("./output").rglob("*.json"):
    data = json.load(jf.open())
    if not data.get("records"):
        continue
    httpx.post(
        API_URL,
        headers={"Authorization": f"Bearer {API_KEY}"},
        json=data,
        timeout=30,
    )

Webhook / Event Streaming

Pipe probe results to a message broker:

"""Emit probe results as Kafka messages."""
import json
from pathlib import Path
from kafka import KafkaProducer

producer = KafkaProducer(
    bootstrap_servers="kafka:9092",
    value_serializer=lambda v: json.dumps(v).encode(),
)

for jf in Path("./output").rglob("*.json"):
    data = json.load(jf.open())
    producer.send("metadata-events", value=data)
producer.flush()