https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I am linking my flows to datahub, is ...
# ask-marvin
y
@Marvin I am linking my flows to datahub, is there any way to alway emit_flow, no matter what's the run result of the run
Copy code
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()
m
✅ thought for 295.0 seconds
Yes — in Prefect 3.x the best way is to emit from flow state hooks, not from the end of the flow body. If you call 
datahub_emitter.emit_flow()
at the bottom of the flow, it will be skipped whenever the flow errors before reaching that line. Use the flow hooks: - 
on_completion
- 
on_failure
- optionally 
on_crashed
- optionally 
on_cancellation
Example:
Copy code
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

def emit_datahub(flow, flow_run, state):
    try:
        datahub_emitter.emit_flow()
    except Exception as exc:
        # optional: log this somewhere
        print(f"Failed to emit DataHub metadata: {exc}")

@flow(
    name="ETL flow",
    description="Extract transform load flow",
    on_completion=[emit_datahub],
    on_failure=[emit_datahub],
    on_crashed=[emit_datahub],
    on_cancellation=[emit_datahub],
)
def etl():
    data = transform("This is data")
    return data
A couple of notes: - 
add_task(...)
should still happen inside your tasks, as you’re already doing. - 
emit_flow()
should not be relied on at the end of the flow body if you need it to happen on failures too. - Wrapping 
emit_flow()
in 
try/except
inside the hook is a good idea so DataHub emission problems do not mask the real flow result. If you want, I can also show a version that only emits on 
completed
+ 
failed
, or help verify this against the exact 
prefect-datahub
version you’re using.
y
@Marvin can it be triggered also when the flow is in long-running (like showing running first and then finish on the datahub UI)
m
✅ thought for 268.5 seconds
Short answer: not cleanly with the current 
prefect-datahub
implementation. You can trigger something when the flow enters 
RUNNING
using Prefect’s 
on_running
hook, but the current 
DatahubEmitter.emit_flow()
implementation does not appear to fully model flow lifecycle transitions like “running now, finished later” for flows. So the answer is: - Prefect side: yes, you can hook into 
RUNNING
- DataHub emitter side: likely not enough by itself to show a proper running → finished transition in DataHub UI for flow runs Example of the Prefect hook:
Copy code
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
def transform(data):
    datahub_emitter.add_task(
        inputs=[Dataset("snowflake", "mydb.schema.tableA")],
        outputs=[Dataset("snowflake", "mydb.schema.tableC")],
    )
    return data.split(" ")

def emit_running(flow, flow_run, state):
    datahub_emitter.emit_flow()

def emit_terminal(flow, flow_run, state):
    datahub_emitter.emit_flow()

@flow(
    name="ETL flow",
    on_running=[emit_running],
    on_completion=[emit_terminal],
    on_failure=[emit_terminal],
    on_crashed=[emit_terminal],
    on_cancellation=[emit_terminal],
)
def etl():
    return transform("This is data")
But a few important caveats: - This may emit multiple times, but that does not necessarily mean DataHub will render it as “running first, then completed/failed”. - From the current integration behavior, 
emit_flow()
seems oriented around emitting the flow metadata/run event, but not a full state-transition lifecycle for flows the same way you’re hoping. - So calling it in 
on_running
and again at the end may create duplicate emissions rather than a nice status progression in the UI. If your goal is specifically: “I want DataHub UI to show a flow run as RUNNING while it executes, then COMPLETED/FAILED when done” then this likely requires an enhancement to 
prefect-datahub
, not just a different Prefect hook setup. Useful Prefect docs on hooks: - State change hooks - Flow API If you want, I can help you with either: 1. a best-effort hook pattern for today, or 2. a small custom emitter patch to support running/final state emission more explicitly.
y
Is it possible to use it in prefect automation? @Marvin
m
✅ thought for 136.6 seconds
Yes, but not directly. Prefect Automations cannot run arbitrary Python like 
DatahubEmitter.emit_flow()
. So you cannot use an automation to directly call the DataHub emitter inside your flow environment. What automations can do: - trigger on flow run events like: - 
prefect.flow-run.running
- 
prefect.flow-run.completed
- 
prefect.flow-run.failed
- then execute supported actions such as: - 
CallWebhook
- notifications - flow/deployment control actions So the workable pattern is: 1. Prefect Automation watches flow state changes 2. Automation sends a webhook 3. Your webhook service calls DataHub So for your use case: - If you want “running” and “finished” events: yes, automations can trigger on those states - If you want to call 
DatahubEmitter.emit_flow()
itself: no, not from the automation alone A typical setup would be:
Copy code
text
Prefect flow-run event
-> Automation
-> Webhook
-> Your service
-> DataHub API / custom emitter logic
This is useful if you want DataHub updates without modifying flow code, but there’s a tradeoff: - automations know about flow state changes - they do not automatically have the same in-process lineage context that 
prefect-datahub
collects from 
add_task(...)
So if your lineage depends on:
Copy code
python
datahub_emitter.add_task(inputs=[...], outputs=[...])
then doing everything purely from automations may not reproduce the same rich lineage data unless your webhook service reconstructs it some other way. Recommendation: - Use flow hooks if you want to preserve 
prefect-datahub
task/dataset lineage behavior - Use automations + webhook if your goal is mostly status syncing to DataHub (
running
, 
completed
, 
failed
) Relevant docs: - Automations - Events - Automation actions If you want, I can draft: 1. a Prefect Automation webhook payload for flow state changes, or 2. a recommended hook-based approach for 
prefect-datahub
lineage plus status emission.
3 Views
Open in Slack
PreviousNext
Powered by