You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "mobuchowski (via GitHub)" <gi...@apache.org> on 2025/04/14 16:51:25 UTC
[PR] add root parent information to OpenLineage events [airflow]
mobuchowski opened a new pull request, #49237:
URL: https://github.com/apache/airflow/pull/49237
This PR extends OpenLineage integration in Airflow to support root parent lineage metadata via the newly added `root` field in the ParentRunFacet.
Previously, consumers had to recursively traverse parent references to determine root job in lineage hierarchy. With this update, consumers can immediately identify the DAG-level root job and run for any depth of children task.
More on the underlying OpenLineage feature: https://docs.google.com/document/d/1JypxrJc0MkrarzYfMopK-TF8QiPpjPnk2je4mgq9e4k
This updates both `openlineage` provider as few others that generate OL events by themselves.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] add root parent information to OpenLineage events [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski merged PR #49237:
URL: https://github.com/apache/airflow/pull/49237
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] add root parent information to OpenLineage events [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on PR #49237:
URL: https://github.com/apache/airflow/pull/49237#issuecomment-2808759049
@amoghrajesh looking at this right now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] add root parent information to OpenLineage events [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #49237:
URL: https://github.com/apache/airflow/pull/49237#discussion_r2074032253
##########
providers/dbt/cloud/src/airflow/providers/dbt/cloud/utils/openlineage.py:
##########
@@ -48,7 +48,7 @@ def _get_logical_date(task_instance):
return date
-@require_openlineage_version(provider_min_version="2.0.0")
+@require_openlineage_version(provider_min_version="2.3.0")
Review Comment:
Why 2.3.0 here and 2.2.0 in pyproject.toml ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] add root parent information to OpenLineage events [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #49237:
URL: https://github.com/apache/airflow/pull/49237#discussion_r2074034555
##########
providers/amazon/pyproject.toml:
##########
@@ -138,7 +138,7 @@ dependencies = [
"apache-airflow-providers-mongo"
]
"openlineage" = [
- "apache-airflow-providers-openlineage"
+ "apache-airflow-providers-openlineage>=2.2.0"
Review Comment:
```suggestion
"apache-airflow-providers-openlineage>=2.3.0"
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] add root parent information to OpenLineage events [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #49237:
URL: https://github.com/apache/airflow/pull/49237#discussion_r2055792768
##########
providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py:
##########
@@ -504,11 +502,12 @@ def on_state_change():
adapter_kwargs = {
"run_id": task_uuid,
"job_name": get_job_name(ti),
- "parent_job_name": dagrun.dag_id,
- "parent_run_id": parent_run_id,
"end_time": end_date.isoformat(),
"task": OperatorLineage(),
- "run_facets": get_airflow_debug_facet(),
+ "run_facets": {
+ **get_task_parent_run_id(ti.dag_id, parent_run_id),
Review Comment:
There is validation, I just did not run it with manual method. Added calls to the test now.
##########
providers/openlineage/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -126,6 +127,24 @@ def get_job_name(task: TaskInstance) -> str:
return f"{task.dag_id}.{task.task_id}"
+def get_task_parent_run_id(
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] add root parent information to OpenLineage events [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on PR #49237:
URL: https://github.com/apache/airflow/pull/49237#issuecomment-2808800482
@amoghrajesh rebase fixed those issues.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] add root parent information to OpenLineage events [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #49237:
URL: https://github.com/apache/airflow/pull/49237#discussion_r2055793171
##########
providers/openlineage/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -126,6 +127,24 @@ def get_job_name(task: TaskInstance) -> str:
return f"{task.dag_id}.{task.task_id}"
+def get_task_parent_run_id(
+ parent_run_id: str, parent_job_name: str, parent_job_namespace: str = conf.namespace()
+) -> dict[str, Any]:
+ # Gets parent run id for TASK level events
+ # Those right now will always point to the DAG level run it and name
+ # As we don't handle dags runs from external events yet
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] add root parent information to OpenLineage events [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #49237:
URL: https://github.com/apache/airflow/pull/49237#discussion_r2055794054
##########
providers/openlineage/tests/unit/openlineage/plugins/test_execution.py:
##########
@@ -99,22 +96,15 @@ def setup_job(self, task_name, run_id):
dag = dagbag.dags.get("test_openlineage_execution")
task = dag.get_task(task_name)
- if AIRFLOW_V_3_0_PLUS:
- dagrun_kwargs = {
- "logical_date": DEFAULT_DATE,
- "run_after": DEFAULT_DATE,
- "triggered_by": DagRunTriggeredByType.TEST,
- }
- else:
- dagrun_kwargs = {"execution_date": DEFAULT_DATE}
Review Comment:
The code is running only for Airflow 2, there's no point of having it there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] add root parent information to OpenLineage events [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on PR #49237:
URL: https://github.com/apache/airflow/pull/49237#issuecomment-2827364084
@amoghrajesh want to take another look at that one? @kacpermuda reviewed it and approved.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] add root parent information to OpenLineage events [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on PR #49237:
URL: https://github.com/apache/airflow/pull/49237#issuecomment-2808514018
@mobuchowski test failures on this one. Is it important before today's provider wave?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] add root parent information to OpenLineage events [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #49237:
URL: https://github.com/apache/airflow/pull/49237#discussion_r2074074584
##########
providers/dbt/cloud/src/airflow/providers/dbt/cloud/utils/openlineage.py:
##########
@@ -48,7 +48,7 @@ def _get_logical_date(task_instance):
return date
-@require_openlineage_version(provider_min_version="2.0.0")
+@require_openlineage_version(provider_min_version="2.3.0")
Review Comment:
Fixed. There was a release in the meantime, did not catch all the places.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] add root parent information to OpenLineage events [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #49237:
URL: https://github.com/apache/airflow/pull/49237#discussion_r2074074891
##########
providers/amazon/pyproject.toml:
##########
@@ -138,7 +138,7 @@ dependencies = [
"apache-airflow-providers-mongo"
]
"openlineage" = [
- "apache-airflow-providers-openlineage"
+ "apache-airflow-providers-openlineage>=2.2.0"
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] add root parent information to OpenLineage events [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on PR #49237:
URL: https://github.com/apache/airflow/pull/49237#issuecomment-2851950047
@potiuk done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] add root parent information to OpenLineage events [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #49237:
URL: https://github.com/apache/airflow/pull/49237#discussion_r2050421329
##########
providers/openlineage/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -126,6 +127,24 @@ def get_job_name(task: TaskInstance) -> str:
return f"{task.dag_id}.{task.task_id}"
+def get_task_parent_run_id(
Review Comment:
```suggestion
def get_task_parent_run_facet(
```
##########
providers/openlineage/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -126,6 +127,24 @@ def get_job_name(task: TaskInstance) -> str:
return f"{task.dag_id}.{task.task_id}"
+def get_task_parent_run_id(
+ parent_run_id: str, parent_job_name: str, parent_job_namespace: str = conf.namespace()
+) -> dict[str, Any]:
+ # Gets parent run id for TASK level events
+ # Those right now will always point to the DAG level run it and name
+ # As we don't handle dags runs from external events yet
Review Comment:
Maybe let's do it as proper docstring? I think we can also be more descriptive here, that parent information will be used as root parent information, as for now we do not track DAG parents, so there is no "recursive" check.
##########
providers/openlineage/tests/unit/openlineage/plugins/test_adapter.py:
##########
@@ -637,7 +665,17 @@ def test_emit_dag_started_event(mock_stats_incr, mock_stats_timer, generate_stat
nominal_end_time=event_time.isoformat(),
owners=["airflow"],
description=dag.description,
- run_facets={"airflowDagRun": dag_run_facet},
+ run_facets={
+ "parent": parent_run.ParentRunFacet(
+ run=parent_run.Run(runId=random_uuid),
+ job=parent_run.Job(namespace=namespace(), name="parent_job_name"),
+ root=parent_run.Root(
+ run=parent_run.RootRun(runId=random_uuid),
+ job=parent_run.RootJob(namespace=namespace(), name="parent_job_name"),
+ ),
+ ),
+ "airflowDagRun": dag_run_facet,
+ },
Review Comment:
If we're testing it for dag_start, let's add it to dag_success and dag_failed as well. It's not really testing much, just that run facets passed to the method are included in event, but still worth it probably.
##########
providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py:
##########
@@ -504,11 +502,12 @@ def on_state_change():
adapter_kwargs = {
"run_id": task_uuid,
"job_name": get_job_name(ti),
- "parent_job_name": dagrun.dag_id,
- "parent_run_id": parent_run_id,
"end_time": end_date.isoformat(),
"task": OperatorLineage(),
- "run_facets": get_airflow_debug_facet(),
+ "run_facets": {
+ **get_task_parent_run_id(ti.dag_id, parent_run_id),
Review Comment:
```suggestion
**get_task_parent_run_id(parent_run_id=parent_run_id, parent_job_name=ti.dag_id),
```
I think the order here was wrong, so it would not work. Maybe some uuid validation should be added to catch those kind of errors or we should only allow kwargs?
##########
providers/openlineage/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -126,6 +127,24 @@ def get_job_name(task: TaskInstance) -> str:
return f"{task.dag_id}.{task.task_id}"
+def get_task_parent_run_id(
Review Comment:
Let's be consistent in naming with other utils functions that return facets.
##########
providers/openlineage/src/airflow/providers/openlineage/plugins/macros.py:
##########
@@ -101,3 +87,44 @@ def lineage_parent_id(task_instance: TaskInstance):
lineage_run_id(task_instance),
)
)
+
+
+def lineage_root_parent_id(task_instance: TaskInstance):
+ return "/".join(
+ (
+ lineage_job_namespace(),
+ lineage_root_job_name(task_instance),
+ lineage_root_run_id(task_instance),
+ )
+ )
+
+
+def lineage_root_job_name(task_instance: TaskInstance):
+ return task_instance.dag_id
+
+
+def lineage_root_run_id(task_instance: TaskInstance):
+ return OpenLineageAdapter.build_dag_run_id(
+ dag_id=task_instance.dag_id,
+ logical_date=_get_logical_date(task_instance),
+ clear_number=task_instance.dag_run.clear_number,
+ )
Review Comment:
How do you think we'll make it work once we add parent run facets to DAGs as well (f.e. when DAG is triggered by task from other dag)? Can we already adjust for it somehow to avoid changing the macros later on?
##########
providers/dbt/cloud/src/airflow/providers/dbt/cloud/utils/openlineage.py:
##########
@@ -148,10 +148,19 @@ async def get_artifacts_for_steps(steps, artifacts):
map_index=task_instance.map_index,
)
+ root_parent_run_id = OpenLineageAdapter.build_dag_run_id(
+ dag_id=task_instance.dag_id,
+ logical_date=_get_logical_date(task_instance),
+ clear_number=task_instance.dag_run.clear_number,
+ )
+
parent_job = ParentRunMetadata(
run_id=parent_run_id,
job_name=f"{task_instance.dag_id}.{task_instance.task_id}",
job_namespace=namespace(),
+ root_parent_run_id=root_parent_run_id,
+ root_parent_job_name=task_instance.dag_id,
+ root_parent_job_namespace=namespace(),
Review Comment:
I think we need to bump the OL provider version in decorator for this function to 2.2.0, otherwise passing root_* args here will cause error as we are not sure if OL commmon is >= 1.31.0. It should be:
`@require_openlineage_version(provider_min_version="2.2.0")`
##########
providers/openlineage/tests/unit/openlineage/plugins/test_execution.py:
##########
@@ -99,22 +96,15 @@ def setup_job(self, task_name, run_id):
dag = dagbag.dags.get("test_openlineage_execution")
task = dag.get_task(task_name)
- if AIRFLOW_V_3_0_PLUS:
- dagrun_kwargs = {
- "logical_date": DEFAULT_DATE,
- "run_after": DEFAULT_DATE,
- "triggered_by": DagRunTriggeredByType.TEST,
- }
- else:
- dagrun_kwargs = {"execution_date": DEFAULT_DATE}
Review Comment:
Why is this change needed? Seems unrelated. Did something change in tests?
##########
providers/snowflake/src/airflow/providers/snowflake/utils/openlineage.py:
##########
@@ -154,13 +164,21 @@ def _get_parent_run_facet(task_instance):
from airflow.providers.openlineage.conf import namespace
parent_run_id = _get_ol_run_id(task_instance)
+ root_parent_run_id = _get_ol_dag_run_id(task_instance)
return parent_run.ParentRunFacet(
run=parent_run.Run(runId=parent_run_id),
job=parent_run.Job(
namespace=namespace(),
name=f"{task_instance.dag_id}.{task_instance.task_id}",
),
+ root=parent_run.Root(
+ run=parent_run.RootRun(runId=root_parent_run_id),
+ job=parent_run.RootJob(
+ name=task_instance.dag_id,
+ namespace=namespace(),
+ ),
+ ),
Review Comment:
Same as in dbt provider, I think we need to bump the OL provider requirement in decorator of the main function.
`@require_openlineage_version(provider_min_version="2.2.0")`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] add root parent information to OpenLineage events [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #49237:
URL: https://github.com/apache/airflow/pull/49237#discussion_r2058157248
##########
providers/openlineage/src/airflow/providers/openlineage/plugins/macros.py:
##########
@@ -101,3 +87,44 @@ def lineage_parent_id(task_instance: TaskInstance):
lineage_run_id(task_instance),
)
)
+
+
+def lineage_root_parent_id(task_instance: TaskInstance):
+ return "/".join(
+ (
+ lineage_job_namespace(),
+ lineage_root_job_name(task_instance),
+ lineage_root_run_id(task_instance),
+ )
+ )
+
+
+def lineage_root_job_name(task_instance: TaskInstance):
+ return task_instance.dag_id
+
+
+def lineage_root_run_id(task_instance: TaskInstance):
+ return OpenLineageAdapter.build_dag_run_id(
+ dag_id=task_instance.dag_id,
+ logical_date=_get_logical_date(task_instance),
+ clear_number=task_instance.dag_run.clear_number,
+ )
Review Comment:
As long as we can access it via `TaskInstance` - that would make sense, but is also orthogonal from this PR itself - as we don't block the future solution.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] add root parent information to OpenLineage events [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #49237:
URL: https://github.com/apache/airflow/pull/49237#discussion_r2056548263
##########
providers/openlineage/tests/unit/openlineage/plugins/test_adapter.py:
##########
@@ -637,7 +665,17 @@ def test_emit_dag_started_event(mock_stats_incr, mock_stats_timer, generate_stat
nominal_end_time=event_time.isoformat(),
owners=["airflow"],
description=dag.description,
- run_facets={"airflowDagRun": dag_run_facet},
+ run_facets={
+ "parent": parent_run.ParentRunFacet(
+ run=parent_run.Run(runId=random_uuid),
+ job=parent_run.Job(namespace=namespace(), name="parent_job_name"),
+ root=parent_run.Root(
+ run=parent_run.RootRun(runId=random_uuid),
+ job=parent_run.RootJob(namespace=namespace(), name="parent_job_name"),
+ ),
+ ),
+ "airflowDagRun": dag_run_facet,
+ },
Review Comment:
Done.
##########
providers/snowflake/src/airflow/providers/snowflake/utils/openlineage.py:
##########
@@ -154,13 +164,21 @@ def _get_parent_run_facet(task_instance):
from airflow.providers.openlineage.conf import namespace
parent_run_id = _get_ol_run_id(task_instance)
+ root_parent_run_id = _get_ol_dag_run_id(task_instance)
return parent_run.ParentRunFacet(
run=parent_run.Run(runId=parent_run_id),
job=parent_run.Job(
namespace=namespace(),
name=f"{task_instance.dag_id}.{task_instance.task_id}",
),
+ root=parent_run.Root(
+ run=parent_run.RootRun(runId=root_parent_run_id),
+ job=parent_run.RootJob(
+ name=task_instance.dag_id,
+ namespace=namespace(),
+ ),
+ ),
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] add root parent information to OpenLineage events [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #49237:
URL: https://github.com/apache/airflow/pull/49237#discussion_r2058028094
##########
providers/openlineage/src/airflow/providers/openlineage/plugins/macros.py:
##########
@@ -101,3 +87,44 @@ def lineage_parent_id(task_instance: TaskInstance):
lineage_run_id(task_instance),
)
)
+
+
+def lineage_root_parent_id(task_instance: TaskInstance):
+ return "/".join(
+ (
+ lineage_job_namespace(),
+ lineage_root_job_name(task_instance),
+ lineage_root_run_id(task_instance),
+ )
+ )
+
+
+def lineage_root_job_name(task_instance: TaskInstance):
+ return task_instance.dag_id
+
+
+def lineage_root_run_id(task_instance: TaskInstance):
+ return OpenLineageAdapter.build_dag_run_id(
+ dag_id=task_instance.dag_id,
+ logical_date=_get_logical_date(task_instance),
+ clear_number=task_instance.dag_run.clear_number,
+ )
Review Comment:
I was thinking about using DAGRun.conf to pass some OL information, then we could probably access it from ti level and have access to real root parent id
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] add root parent information to OpenLineage events [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #49237:
URL: https://github.com/apache/airflow/pull/49237#issuecomment-2851523135
needs conflict resolving
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] add root parent information to OpenLineage events [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #49237:
URL: https://github.com/apache/airflow/pull/49237#discussion_r2055797306
##########
providers/openlineage/src/airflow/providers/openlineage/plugins/macros.py:
##########
@@ -101,3 +87,44 @@ def lineage_parent_id(task_instance: TaskInstance):
lineage_run_id(task_instance),
)
)
+
+
+def lineage_root_parent_id(task_instance: TaskInstance):
+ return "/".join(
+ (
+ lineage_job_namespace(),
+ lineage_root_job_name(task_instance),
+ lineage_root_run_id(task_instance),
+ )
+ )
+
+
+def lineage_root_job_name(task_instance: TaskInstance):
+ return task_instance.dag_id
+
+
+def lineage_root_run_id(task_instance: TaskInstance):
+ return OpenLineageAdapter.build_dag_run_id(
+ dag_id=task_instance.dag_id,
+ logical_date=_get_logical_date(task_instance),
+ clear_number=task_instance.dag_run.clear_number,
+ )
Review Comment:
There would need to be a mechanism to propagate this information to a task - which there is not right now.
I don't know how would it work, so I think it's futile to try to fit this in right now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] add root parent information to OpenLineage events [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #49237:
URL: https://github.com/apache/airflow/pull/49237#discussion_r2055798460
##########
providers/dbt/cloud/src/airflow/providers/dbt/cloud/utils/openlineage.py:
##########
@@ -148,10 +148,19 @@ async def get_artifacts_for_steps(steps, artifacts):
map_index=task_instance.map_index,
)
+ root_parent_run_id = OpenLineageAdapter.build_dag_run_id(
+ dag_id=task_instance.dag_id,
+ logical_date=_get_logical_date(task_instance),
+ clear_number=task_instance.dag_run.clear_number,
+ )
+
parent_job = ParentRunMetadata(
run_id=parent_run_id,
job_name=f"{task_instance.dag_id}.{task_instance.task_id}",
job_namespace=namespace(),
+ root_parent_run_id=root_parent_run_id,
+ root_parent_job_name=task_instance.dag_id,
+ root_parent_job_namespace=namespace(),
Review Comment:
Updated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org