Add OpenLineage integration#22475
Conversation
Signed-off-by: merobi-hub <merobi@gmail.com>
Signed-off-by: merobi-hub <merobi@gmail.com>
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
|
|
||
| producer = facet._get_producer() | ||
| assert producer.startswith("https://") | ||
| assert "github.com" in producer |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 077d855556
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| class PrefectOpenLineageListener: | ||
| def __init__( | ||
| self, | ||
| client: client | None = None, |
There was a problem hiding this comment.
On supported Python 3.10-3.13, importing prefect_openlineage.listener evaluates this annotation immediately; from prefect import client binds the prefect.client module, and client | None raises TypeError: unsupported operand type(s) for |. This prevents the listener module and its tests from importing before any events can be processed; use the actual client type, Any, or postponed/string annotations.
Useful? React with 👍 / 👎.
| flow_run_name = event.resource.name | ||
| flow_name = res["prefect.resource.name"] | ||
|
|
||
| if flow_run_name: |
There was a problem hiding this comment.
Because this condition is inside the for event.related loop and stays true after the flow related resource is seen, a normal deployment-backed flow event with related flow/deployment/work-queue/work-pool/tag resources emits the same OpenLineage flow event once for every remaining related item. This duplicates every flow START/COMPLETE/FAIL record; process only the matching flow resource or move the emission after the loop.
Useful? React with 👍 / 👎.
| [project] | ||
| dependencies = [ | ||
| "prefect>=3.7.0", | ||
| "openlineage-python>=1.0.0", |
There was a problem hiding this comment.
The declared range lets resolvers keep openlineage-python==1.0.0, but this integration imports openlineage.client.event_v2 and openlineage.client.facet_v2; those modules are not present in the 1.0.0 client, so environments with an older 1.x pin can install successfully and then fail at import. Please raise the lower bound to the first release containing these modules.
Useful? React with 👍 / 👎.
| ) | ||
|
|
||
| async def collect_and_process_task_runs(self, prefect_version: str, event: Event, event_state: str) -> None: | ||
| task_name = event.resource.name.split("-")[0] |
There was a problem hiding this comment.
When a task name itself contains hyphens, Prefect task run names are still suffixed with -<dynamic-key>, so splitting on the first hyphen turns a task like extract-users-abc into the OpenLineage job name/run-id seed extract. This collapses distinct hyphenated tasks into the same lineage job prefix; strip only the run suffix from the right or use the task metadata in the payload instead.
Useful? React with 👍 / 👎.
Merging this PR will not alter performance
Comparing 
Footnotes
|
Signed-off-by: merobi-hub <merobi@gmail.com>
Signed-off-by: merobi-hub <merobi@gmail.com>
Signed-off-by: merobi-hub <merobi@gmail.com>
Signed-off-by: merobi-hub <merobi@gmail.com>
Signed-off-by: merobi-hub <merobi@gmail.com>
Signed-off-by: merobi-hub <merobi@gmail.com>
Signed-off-by: merobi-hub <merobi@gmail.com>
2 participants
This change would add an OpenLineage integration to the project. OpenLineage is a widely adopted open specification for data lineage. It is an intentionally simple JSON spec that is extensible by "facets."
OpenLineage is a natural fit with Prefect in part because it, too, is event-first in design. Although it now supports "static" lineage, its focus has always been on capturing and emitting information at runtime. Another reason OpenLineage is a good fit is that it offers a robust Python client.
This integration makes use of the OpenLineage Python client to:
Start,Complete,Fail)This PR would add:
get_events_subscriberclient and listens forRunning,CompletedandFailedevents from Prefect tasks and flowsPrefectDeploymentRunFacet, that captures information about deployments for inclusion in task and flow run eventsThe integration has been tested and validated locally. The consumer used in testing was Marquez.
A sample OpenLineage start event:
{ "eventTime": "2026-06-24T10:39:45.415Z", "eventType": "START", "inputs": [ { "facets": {}, "name": "customers", "namespace": "duckdb:///customers_db" } ], "job": { "facets": { "jobType": { "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.46.0/client/python", "_schemaURL": "https://raw.githubusercontent.com/OpenLineage/OpenLineage/main/spec/OpenLineage.json#/definitions/JobTypeJobFacet", "integration": "Prefect", "jobType": "TASK", "processingType": "BATCH" } }, "name": "fake_ingest", "namespace": "prefect_test" }, "outputs": [], "producer": "https://github.com/OpenLineage/OpenLineage/tree/$VERSION/integration/prefect", "run": { "facets": { "jobDependencies": { "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.46.0/client/python", "_schemaURL": "https://openlineage.io/spec/facets/1-0-1/JobDependenciesRunFacet.json#/$defs/JobDependenciesRunFacet", "downstream": [], "upstream": [ { "job": { "name": "get_stars", "namespace": "prefect_test" } } ] }, "nominalTime": { "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.46.0/client/python", "_schemaURL": "https://raw.githubusercontent.com/OpenLineage/OpenLineage/main/spec/OpenLineage.json#/definitions/NominalTimeRunFacet", "nominalStartTime": "2026-06-24T10:39:45.414730Z" }, "parentRun": { "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.46.0/client/python", "_schemaURL": "https://raw.githubusercontent.com/OpenLineage/OpenLineage/main/spec/OpenLineage.json#/definitions/ParentRunFacet", "job": { "name": "GitHub Stars", "namespace": "prefect_test" }, "run": { "runId": "019ef936-e652-7116-a6b2-f75027f79e20" } }, "prefectDeployment": { "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.46.0/client/python", "_schemaURL": "https://raw.githubusercontent.com/OpenLineage/OpenLineage/main/spec/OpenLineage.json#/definitions/BaseFacet", "created": "2026-04-29T01:55:23.081088+00:00", "deployment_id": "1ace9016-b961-4c8f-bd93-412e565ed1a5", "name": "flaky_deploy", "updated": "2026-06-04T22:49:52.551000+00:00" }, "processingEngine": { "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.46.0/client/python", "_schemaURL": "https://openlineage.io/spec/facets/1-1-1/ProcessingEngineRunFacet.json#/$defs/ProcessingEngineRunFacet", "name": "Prefect", "version": "3.6.26" }, "tags": { "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.46.0/client/python", "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/TagsRunFacet.json#/$defs/TagsRunFacet", "tags": [ { "key": "openlineage_client_version", "source": "OPENLINEAGE_CLIENT", "value": "1.46.0" } ] } }, "runId": "019ef936-e707-71c7-a0cc-91c0095de3e6" }, "schemaURL": "https://openlineage.io/spec/1-0-5/OpenLineage.json#/definitions/RunEvent" }A sample complete event:
{ "eventTime": "2026-06-24T10:39:45.490Z", "eventType": "COMPLETE", "inputs": [ { "facets": {}, "name": "customers", "namespace": "duckdb:///customers_db" } ], "job": { "facets": { "jobType": { "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.46.0/client/python", "_schemaURL": "https://raw.githubusercontent.com/OpenLineage/OpenLineage/main/spec/OpenLineage.json#/definitions/JobTypeJobFacet", "integration": "Prefect", "jobType": "TASK", "processingType": "BATCH" } }, "name": "fake_ingest", "namespace": "prefect_test" }, "outputs": [], "producer": "https://github.com/OpenLineage/OpenLineage/tree/$VERSION/integration/prefect", "run": { "facets": { "jobDependencies": { "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.46.0/client/python", "_schemaURL": "https://openlineage.io/spec/facets/1-0-1/JobDependenciesRunFacet.json#/$defs/JobDependenciesRunFacet", "downstream": [], "upstream": [ { "job": { "name": "get_stars", "namespace": "prefect_test" } } ] }, "nominalTime": { "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.46.0/client/python", "_schemaURL": "https://raw.githubusercontent.com/OpenLineage/OpenLineage/main/spec/OpenLineage.json#/definitions/NominalTimeRunFacet", "nominalStartTime": "2026-06-24T10:39:45.414730Z" }, "parentRun": { "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.46.0/client/python", "_schemaURL": "https://raw.githubusercontent.com/OpenLineage/OpenLineage/main/spec/OpenLineage.json#/definitions/ParentRunFacet", "job": { "name": "GitHub Stars", "namespace": "prefect_test" }, "run": { "runId": "019ef936-e652-7116-a6b2-f75027f79e20" } }, "prefectDeployment": { "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.46.0/client/python", "_schemaURL": "https://raw.githubusercontent.com/OpenLineage/OpenLineage/main/spec/OpenLineage.json#/definitions/BaseFacet", "created": "2026-04-29T01:55:23.081088+00:00", "deployment_id": "1ace9016-b961-4c8f-bd93-412e565ed1a5", "name": "flaky_deploy", "updated": "2026-06-04T22:49:52.551000+00:00" }, "processingEngine": { "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.46.0/client/python", "_schemaURL": "https://openlineage.io/spec/facets/1-1-1/ProcessingEngineRunFacet.json#/$defs/ProcessingEngineRunFacet", "name": "Prefect", "version": "3.6.26" }, "tags": { "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.46.0/client/python", "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/TagsRunFacet.json#/$defs/TagsRunFacet", "tags": [ { "key": "openlineage_client_version", "source": "OPENLINEAGE_CLIENT", "value": "1.46.0" } ] } }, "runId": "019ef936-e707-71c7-a0cc-91c0095de3e6" }, "schemaURL": "https://openlineage.io/spec/1-0-5/OpenLineage.json#/definitions/RunEvent" }To dos:
Checklist
<link to issue>"mint.json.