Prefect Integration with DataHub

Overview

DataHub supports integration with Prefect, allowing you to ingest:

Prefect flow and task metadata
Flow run and Task run information
Lineage information (when available)

This integration enables you to track and monitor your Prefect workflows within DataHub, providing a comprehensive view of your data pipeline activities.

Prefect DataHub Block

What is a Prefect DataHub Block?

Blocks in Prefect are primitives that enable the storage of configuration and provide an interface for interacting with external systems. The prefect-datahub block uses the DataHub REST emitter to send metadata events while running Prefect flows.

Prerequisites

Use either Prefect Cloud (recommended) or a self-hosted Prefect server.
For Prefect Cloud setup, refer to the Cloud Quickstart guide.
For self-hosted Prefect server setup, refer to the Host Prefect Server guide.
Ensure the Prefect API URL is set correctly. Verify using:
shell
```
prefect profile inspect
```
API URL format:
- Prefect Cloud: https://api.prefect.cloud/api/accounts/<account_id>/workspaces/<workspace_id>
- Self-hosted: http://<host>:<port>/api

Upgrading from Prefect v2

prefect-datahub ≥ 1.0.0 requires Prefect v3 and contains two breaking changes for existing users.

Module path changed. The emitter moved from prefect_datahub.prefect_datahub to prefect_datahub.datahub_emitter. Update any direct imports:

python

# Before (Prefect v2)
from prefect_datahub.prefect_datahub import DatahubEmitter

# After (Prefect v3)
from prefect_datahub.datahub_emitter import DatahubEmitter

Saved blocks must be re-registered. Blocks previously saved to Prefect Cloud or a self-hosted server under the old entry point (prefect_datahub.prefect_datahub:DatahubEmitter) will fail to load after the upgrade because that module no longer exists. Re-register by running .save() again:

python

from prefect_datahub.datahub_emitter import DatahubEmitter

DatahubEmitter(datahub_rest_url="...", env="PROD").save("MY-DATAHUB-BLOCK", overwrite=True)

In Prefect v3, block types are discovered automatically when the collection package is imported — the entry point no longer needs to reference the class explicitly.

Setup Instructions

1. Installation

Install prefect-datahub using pip:

shell

pip install 'prefect-datahub'

Note: Requires Python 3.10+ and Prefect 3.x (>=3.0.0,<4.0.0). Prefect 2.x is not supported. If you are upgrading from Prefect 2.x, upgrade Prefect first and re-register the DataHub block (see Upgrading from Prefect v2).

2. Saving Configurations to a Block

Save your configuration to the Prefect block document store:

python

from prefect_datahub.datahub_emitter import DatahubEmitter

DatahubEmitter(
    datahub_rest_url="http://localhost:8080",
    env="PROD",
    platform_instance="local_prefect"
).save("MY-DATAHUB-BLOCK")

Configuration options:

Config	Type	Default	Description
datahub_rest_url	`str`	`http://localhost:8080`	DataHub GMS REST URL
env	`str`	`PROD`	Environment for assets (see FabricType)
platform_instance	`str`	`None`	Platform instance for assets (see Platform Instances)
emit_mode	`EmitMode`	`ASYNC`	Emit mode for writes to DataHub. `ASYNC` (default) avoids blocking on a synchronous commit per write, reducing GMS load at high volume. Use `SYNC_WAIT`/`SYNC_PRIMARY` for read-after-write or raise-on-failure guarantees.

3. Using the Block in Prefect Workflows

Load and use the saved block in your Prefect workflows:

python

from prefect import flow, task
from prefect_datahub.dataset import Dataset
from prefect_datahub.datahub_emitter import DatahubEmitter

datahub_emitter = DatahubEmitter.load("MY-DATAHUB-BLOCK")

@task(name="Transform", description="Transform the data")
def transform(data):
    data = data.split(" ")
    datahub_emitter.add_task(
        inputs=[Dataset("snowflake", "mydb.schema.tableA")],
        outputs=[Dataset("snowflake", "mydb.schema.tableC")],
    )
    return data

@flow(name="ETL flow", description="Extract transform load flow")
def etl():
    data = transform("This is data")
    datahub_emitter.emit_flow()

Note: To emit tasks, you must call emit_flow(). Otherwise, no metadata will be emitted.

Concept Mapping

Prefect Concept	DataHub Concept
Flow	DataFlow
Flow Run	DataProcessInstance
Task	DataJob
Task Run	DataProcessInstance
Task Tag	Tag

Validation and Troubleshooting

Validating the Setup

Check the Prefect UI's Blocks menu for the DataHub emitter.
Run a Prefect workflow and look for DataHub-related log messages:
text
```
Emitting flow to datahub...
Emitting tasks to datahub...
```

Debugging Common Issues

Incorrect Prefect API URL

If the Prefect API URL is incorrect, set it manually:

shell

prefect config set PREFECT_API_URL='http://127.0.0.1:4200/api'

DataHub Connection Error

If you encounter a ConnectionError: HTTPConnectionPool(host='localhost', port=8080), ensure that your DataHub GMS service is running.

Additional Resources

For more information or support, please refer to the official Prefect and DataHub documentation or reach out to their respective communities.