Custom Integrations

When the Built-In Connectors Are Not Enough

DataHub ships with 80+ ingestion connectors: Snowflake, BigQuery, dbt, Airflow, Looker, and many more
But most organizations also have internal systems that are not covered
Examples: a homegrown key-value config store, a proprietary ETL framework, an internal ML feature store, a CI/CD pipeline that creates derived datasets
Without a custom integration, those systems are invisible to your data catalog
This module teaches you how to make any system visible to DataHub in 30 lines of Python

Four Reasons to Build a Custom Source

Internal system not covered by built-in connectors — homegrown config store, proprietary ETL engine
Emit metadata from a CI/CD pipeline step — mark a dataset "built from branch X" at deploy time
Build an internal tool that creates DataHub assets programmatically — a self-service table registration UI for data engineers
Emit lineage from a custom pipeline that is not Airflow or dbt — a Spark job that reads from S3 and writes to Redshift
You do not need to build a full structured connector for all of these
For scripts and one-off jobs: the DataHub REST emitter is enough
For reusable connectors: the Source base class is the right structure

The DataHub Python SDK: Two Ways to Emit

Install: pip install acryl-datahub
Option 1: DatahubRestEmitter (production) — connects to DataHub's REST API, emits MetadataChangeProposals (MCPs) directly to GMS
Use this for scripts, pipelines, and tools that run against a live DataHub instance
Option 2: SynchronizedFileEmitter (debugging) — writes MCPs to a local JSON file instead of a live server; inspect the file, then switch to the REST emitter once the output looks right

MCPs: The Unit of Metadata Change

MCP = MetadataChangeProposal
An MCP says exactly one thing: "for this entity URN, set this aspect to this value"
entityUrn — the unique identifier for the entity being updated
entityType — the type of entity: dataset, dashboard, dataJob, etc.
aspectName — which aspect you are updating: schemaMetadata, datasetProperties, ownership, etc.
aspect — the new value of that aspect
changeType — the operation: UPSERT (create or update) or DELETE

URNs: How DataHub Identifies Every Entity

Recall from Core Concepts: every entity has exactly one URN and it never changes
As a developer, you will construct and read URNs directly
Dataset: urn:li:dataset:(urn:li:dataPlatform:PLATFORM,TABLE_NAME,ENV)
Dashboard: urn:li:dashboard:(PLATFORM,DASHBOARD_ID)
User: urn:li:corpuser:email@company.com
Real examples: urn:li:dataset:(urn:li:dataPlatform:snowflake,PROD.FINANCE.FACT_REVENUE_DAILY,PROD), urn:li:dataset:(urn:li:dataPlatform:internaldb,mydb.users,PROD)
Tip: copy URNs directly from the browser URL bar on any entity page

Map Your Entities Before Writing Any Code

"Before you write a single line of Python, map your source system's objects to DataHub entity types"
Table, view, file → dataset
BI report or chart → chart or dashboard
ETL pipeline or job → dataJob
Feature store feature group → mlFeatureTable
Common mistake: using the wrong entity type (e.g. dashboard for a table) creates an entity that looks broken in the UI and is hard to fix after 500 entities have been ingested

Emitting Lineage: Connecting Your Assets to the Broader Graph

UpstreamLineage declares that one dataset is derived from another.

from datahub.emitter.mcp import MetadataChangeProposalWrapper
from datahub.emitter.rest_emitter import DatahubRestEmitter
from datahub.metadata.schema_classes import (
    UpstreamLineageClass, UpstreamClass, DatasetLineageTypeClass
)

upstream_urn = "urn:li:dataset:(urn:li:dataPlatform:snowflake,PROD.CORE.raw_events,PROD)"
downstream_urn = "urn:li:dataset:(urn:li:dataPlatform:internaldb,mydb.aggregated_metrics,PROD)"

lineage_mcp = MetadataChangeProposalWrapper(
    entityUrn=downstream_urn,
    aspect=UpstreamLineageClass(
        upstreams=[UpstreamClass(dataset=upstream_urn, type=DatasetLineageTypeClass.TRANSFORMED)]
    ),
)

emitter = DatahubRestEmitter("http://localhost:8080")
emitter.emit(lineage_mcp)

Lineage is what turns a flat list of tables into a graph. Without lineage, your custom integration gives DataHub a list of entities. With lineage, DataHub knows how those entities relate to everything else in the catalog. Walk through the code carefully: the lineage MCP is emitted on the downstream entity (the one that was derived), and it points to the upstream. That direction is important and counterintuitive for some participants who expect to emit it the other way around. Also call out the type field: TRANSFORMED is the right choice for most data pipeline scenarios where data was filtered, aggregated, or joined. The distinction between COPY and TRANSFORMED matters for impact analysis: a TRANSFORMED relationship means downstream data may differ from upstream, which is a critical signal during incident triage.

MCEs Are the Old Format. Use MCPs.

"Always use MetadataChangeProposals (MCPs) for new integrations. Never use MCEs."
MCP (MetadataChangeProposal): current standard
MCE (MetadataChangeEvent): legacy
Why MCEs still exist: old connectors and documentation may still reference them; they still work for backward compatibility; some blog posts and StackOverflow answers show MCE examples from 2021–2022

Validate Your Connection Before Emitting Anything

"Run datahub check server-config before starting any ingestion script"
Uses the host and token already configured via datahub init (or the DATAHUB_GMS_URL / DATAHUB_GMS_TOKEN env vars)
Validates that the DataHub GMS endpoint is reachable
Confirms your authentication token is valid
Fails fast before you emit a single entity

Brief: Screenshot of a terminal window running datahub check server-config with the server config printed to output, confirming GMS is reachable and the token is valid.

Structured Ingestion Sources: When to Go Further

Two levels of custom integration: a script with DatahubRestEmitter (one-off jobs, CI/CD steps, quick scripts), or the Source base class (a reusable connector teams configure with a recipe YAML)
When to use the Source base class: multiple teams or projects will run this connector
You want configuration validation at recipe parse time
You need state management (e.g. only ingest entities modified since the last run)

Build a Custom Ingestion Script for InternalDB

Scenario: InternalDB is a mock internal system with 5 tables. DataHub has no connector for it. You will build the ingestion script.
internaldb.users (user_id, email, created_at)
internaldb.sessions (session_id, user_id, started_at, duration_seconds)
internaldb.events (event_id, session_id, event_type, occurred_at)
internaldb.aggregated_metrics (metric_name, metric_value, computed_at) — downstream of snowflake.PROD.CORE.raw_events
internaldb.revenue_summary (period, revenue, region) — downstream of snowflake.PROD.FINANCE.raw_revenue

Brief: Screenshot of the lab handout card listing the 5 InternalDB tables and their columns, as distributed to participants at their table.

Read the spec aloud once before releasing participants to work. Emphasize the attempt-first structure: the goal is to try to build the script from memory and from the patterns on the previous slides before opening the solution. Eight minutes of genuine struggle is more valuable than copying the solution immediately. Walk the room during the lab. Look for the two most common failure modes: wrong URN format (forgetting the inner urn:li:dataPlatform: wrapper) and incorrect GMS host. If someone is stuck for more than a few minutes on a connection error, send them to the "Validate Your Connection Before Emitting Anything" PRO TIP slide a few slides back. When participants succeed, ask them to also verify lineage: navigate to internaldb.aggregated_metrics in DataHub and confirm the upstream Snowflake table appears in the lineage graph, not just in the entity list.

Check for Understanding

You want to emit metadata from a nightly batch job that runs in your internal ETL framework. The job reads from an S3 bucket and writes to a Redshift table. Which DataHub entity types would you create, and which aspect would you use to connect them?
Your colleague finds a blog post from 2021 showing how to emit metadata using MetadataChangeEvent. Should they follow that example? Why or why not?
Your script runs without any Python errors, but when you search DataHub the tables are not there. What is the first thing you check?

Summary + What's Next

When to build a custom integration: internal systems, CI/CD pipelines, custom lineage, programmatic asset creation
The DataHub Python SDK: DatahubRestEmitter for production, file emitter for debugging
MCPs: the unit of metadata change, with fields entityUrn, entityType, aspectName, and aspect
URN format for datasets: urn:li:dataset:(urn:li:dataPlatform:PLATFORM,TABLE_NAME,ENV)
Emitting lineage with UpstreamLineageClass
When to use a plain script vs. the Source base class