You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "dolfinus (via GitHub)" <gi...@apache.org> on 2024/07/18 10:47:19 UTC
[PR] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
dolfinus opened a new pull request, #40854:
URL: https://github.com/apache/airflow/pull/40854
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
closes: #40798
* Make `AirflowRunFacet` attributes `task`, `taskInstance` and `taskUuid` optional.
* Add the same facet to DAG run events with eventType=STARTED.
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1684204486
##########
airflow/providers/openlineage/plugins/listener.py:
##########
@@ -420,7 +421,8 @@ def on_dag_run_running(self, dag_run: DagRun, msg: str) -> None:
nominal_end_time=data_interval_end,
# AirflowJobFacet should be created outside ProcessPoolExecutor that pickles objects,
# as it causes lack of some TaskGroup attributes and crashes event emission.
- job_facets={**get_airflow_job_facet(dag_run=dag_run)},
+ job_facets=get_airflow_job_facet(dag_run=dag_run),
+ run_facets=get_airflow_dag_run_facet(dag_run),
Review Comment:
I think we want to minimise operations we do in the `on_dag_run_running` in listener, as it's executed before we move to a separate process, so yes, if possible i would move it to adapter.
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on PR #40854:
URL: https://github.com/apache/airflow/pull/40854#issuecomment-2236446811
@dolfinus I think this should be a different facet or added to `AirflowJobFacet` that we attach to DagRun events.
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "dolfinus (via GitHub)" <gi...@apache.org>.
dolfinus commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1684201354
##########
airflow/providers/openlineage/plugins/listener.py:
##########
@@ -420,7 +421,8 @@ def on_dag_run_running(self, dag_run: DagRun, msg: str) -> None:
nominal_end_time=data_interval_end,
# AirflowJobFacet should be created outside ProcessPoolExecutor that pickles objects,
# as it causes lack of some TaskGroup attributes and crashes event emission.
- job_facets={**get_airflow_job_facet(dag_run=dag_run)},
+ job_facets=get_airflow_job_facet(dag_run=dag_run),
+ run_facets=get_airflow_dag_run_facet(dag_run),
Review Comment:
Almost of logic could be moved from on_dag_run_running to dag_started. Is OpenLineageAdapter responsible for building facets in the first place?
--
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] openlineage: Add AirflowDagRunFacet to dag runEvents [airflow]
Posted by "dolfinus (via GitHub)" <gi...@apache.org>.
dolfinus commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1686247089
##########
airflow/providers/openlineage/plugins/adapter.py:
##########
@@ -329,19 +330,26 @@ def dag_started(
self,
dag_run: DagRun,
msg: str,
- nominal_start_time: str,
- nominal_end_time: str,
Review Comment:
Reverted.
But IMHO to be a public api this method should be at least properly documented. And maybe even redesigned, as it currently accepts DagRun, unlike other methods which do not accept Airflow models, but instead set of separated fields to just build OpenLineage events from.
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "dolfinus (via GitHub)" <gi...@apache.org>.
dolfinus commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1684370206
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -307,6 +308,17 @@ class TaskGroupInfo(InfoJsonEncodable):
]
+def get_airflow_dag_run_facet(dag_run: DagRun) -> dict[str, BaseFacet]:
+ if not dag_run.dag:
+ return {}
+ return {
+ "airflow_dagRun": AirflowDagRunFacet(
Review Comment:
Let's use `airflowDagRun` as a key for now, and discuss this in https://github.com/OpenLineage/OpenLineage/issues/2866.
--
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] openlineage: Add AirflowDagRunFacet to dag runEvents [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1686194775
##########
airflow/providers/openlineage/plugins/adapter.py:
##########
@@ -329,19 +330,26 @@ def dag_started(
self,
dag_run: DagRun,
msg: str,
- nominal_start_time: str,
- nominal_end_time: str,
Review Comment:
Instead of building it in listener, we moved it to the inside of the adapter's method. I'm okay with that, but not sure if the adapter methods are part of "public" api and it's `breaking` change @mobuchowski ?
##########
airflow/providers/openlineage/plugins/adapter.py:
##########
@@ -329,19 +330,26 @@ def dag_started(
self,
dag_run: DagRun,
msg: str,
- nominal_start_time: str,
- nominal_end_time: str,
job_facets: dict[str, BaseFacet] | None = None, # Custom job facets
+ run_facets: dict[str, BaseFacet] | None = None, # Custom run facets
):
+ if not run_facets:
+ run_facets = {}
Review Comment:
Is this used anywhere now that we don't build this new run facet in listener? I think we can remove it for now if not used.
--
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] openlineage: Add AirflowDagRunFacet to dag runEvents [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1686370968
##########
airflow/providers/openlineage/plugins/adapter.py:
##########
@@ -329,19 +330,26 @@ def dag_started(
self,
dag_run: DagRun,
msg: str,
- nominal_start_time: str,
- nominal_end_time: str,
Review Comment:
@dolfinus it's more likely we'll redesign the whole mechanism in Airflow 3 due to AIP-72 - and possibly stop running on worker altogether. And I think not depending on Airflow models there is a good idea.
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "dolfinus (via GitHub)" <gi...@apache.org>.
dolfinus commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1684201354
##########
airflow/providers/openlineage/plugins/listener.py:
##########
@@ -420,7 +421,8 @@ def on_dag_run_running(self, dag_run: DagRun, msg: str) -> None:
nominal_end_time=data_interval_end,
# AirflowJobFacet should be created outside ProcessPoolExecutor that pickles objects,
# as it causes lack of some TaskGroup attributes and crashes event emission.
- job_facets={**get_airflow_job_facet(dag_run=dag_run)},
+ job_facets=get_airflow_job_facet(dag_run=dag_run),
+ run_facets=get_airflow_dag_run_facet(dag_run),
Review Comment:
Almost of logic could be moved from `on_dag_run_running` to `dag_started`. Is `OpenLineageAdapter` responsible for building facets in the first place?
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1684206014
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -307,6 +308,17 @@ class TaskGroupInfo(InfoJsonEncodable):
]
+def get_airflow_dag_run_facet(dag_run: DagRun) -> dict[str, BaseFacet]:
+ if not dag_run.dag:
+ return {}
+ return {
+ "airflow_dagRun": AirflowDagRunFacet(
Review Comment:
@mobuchowski WDYT? Should we start change current one and start sending duplicate under `airflow_state` key or change the spec and adjust this new one to `airflowDagRun`?
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1684196136
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -307,6 +308,17 @@ class TaskGroupInfo(InfoJsonEncodable):
]
+def get_airflow_dag_run_facet(dag_run: DagRun) -> dict[str, BaseFacet]:
+ if not dag_run.dag:
+ return {}
+ return {
+ "airflow_dagRun": AirflowDagRunFacet(
Review Comment:
Interesting, i missed that part of specification about naming when I created the `airflowState` facet.
##########
airflow/providers/openlineage/plugins/listener.py:
##########
@@ -420,7 +421,8 @@ def on_dag_run_running(self, dag_run: DagRun, msg: str) -> None:
nominal_end_time=data_interval_end,
# AirflowJobFacet should be created outside ProcessPoolExecutor that pickles objects,
# as it causes lack of some TaskGroup attributes and crashes event emission.
- job_facets={**get_airflow_job_facet(dag_run=dag_run)},
+ job_facets=get_airflow_job_facet(dag_run=dag_run),
+ run_facets=get_airflow_dag_run_facet(dag_run),
Review Comment:
Did you check if we can perform this inside the `self.adapter.dag_started` method? It's possible that the pickling problem mentioned above doesn't affect the `dag` and `dagrun` that we use.
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "dolfinus (via GitHub)" <gi...@apache.org>.
dolfinus commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1684319591
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -307,6 +308,17 @@ class TaskGroupInfo(InfoJsonEncodable):
]
+def get_airflow_dag_run_facet(dag_run: DagRun) -> dict[str, BaseFacet]:
+ if not dag_run.dag:
+ return {}
+ return {
+ "airflow_dagRun": AirflowDagRunFacet(
Review Comment:
Is Airflow facets the only exception, or the specification itself should be updated to allow both snake and camel case keys?
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1684212108
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -307,6 +308,17 @@ class TaskGroupInfo(InfoJsonEncodable):
]
+def get_airflow_dag_run_facet(dag_run: DagRun) -> dict[str, BaseFacet]:
+ if not dag_run.dag:
+ return {}
+ return {
+ "airflow_dagRun": AirflowDagRunFacet(
Review Comment:
The way to properly do renaming - assuming we want to - is to publish the facet under both names to let consumers switch to the new facet name.
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "dolfinus (via GitHub)" <gi...@apache.org>.
dolfinus commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1684314367
##########
airflow/providers/openlineage/plugins/listener.py:
##########
@@ -420,7 +421,8 @@ def on_dag_run_running(self, dag_run: DagRun, msg: str) -> None:
nominal_end_time=data_interval_end,
# AirflowJobFacet should be created outside ProcessPoolExecutor that pickles objects,
# as it causes lack of some TaskGroup attributes and crashes event emission.
- job_facets={**get_airflow_job_facet(dag_run=dag_run)},
+ job_facets=get_airflow_job_facet(dag_run=dag_run),
+ run_facets=get_airflow_dag_run_facet(dag_run),
Review Comment:
Moved building run facet and filling up nominal start/end time to the adapter.
Tested on demo dag, don't see any issues with DagRun pickling, unlike TaskGroup. Here is an example of DAG run start event:
<details>
```json
{
"eventTime": "2024-07-19T12:27:26.050677+00:00",
"eventType": "START",
"inputs": [],
"job": {
"facets": {
"airflow": {
"_producer": "https://github.com/apache/airflow/tree/providers-openlineage/1.9.1",
"_schemaURL": "https://raw.githubusercontent.com/OpenLineage/OpenLineage/main/spec/OpenLineage.json#/definitions/BaseFacet",
"taskGroups": {
"section_1": {
"tooltip": "Tasks for section_1",
"ui_color": "CornflowerBlue",
"ui_fgcolor": "#000",
"ui_label": "section_1"
},
"section_2": {
"tooltip": "Tasks for section_2",
"ui_color": "CornflowerBlue",
"ui_fgcolor": "#000",
"ui_label": "section_2"
},
"section_2.inner_section_2": {
"parent_group": "section_2",
"tooltip": "Tasks for inner_section2",
"ui_color": "CornflowerBlue",
"ui_fgcolor": "#000",
"ui_label": "inner_section_2"
}
},
"taskTree": {
"start": {
"section_1.task_1": {
"section_1.task_2": {
"section_2.inner_section_2.task_2": {
"section_2.inner_section_2.task_4": {
"end": {}
}
},
"section_2.inner_section_2.task_3": {
"section_2.inner_section_2.task_4": {
"end": {}
}
},
"section_2.task_1": {
"end": {}
}
},
"section_1.task_3": {
"section_2.inner_section_2.task_2": {
"section_2.inner_section_2.task_4": {
"end": {}
}
},
"section_2.inner_section_2.task_3": {
"section_2.inner_section_2.task_4": {
"end": {}
}
},
"section_2.task_1": {
"end": {}
}
}
}
}
},
"tasks": {
"end": {
"emits_ol_events": false,
"is_setup": false,
"is_teardown": false,
"operator": "airflow.operators.empty.EmptyOperator",
"ui_color": "#e8f7e4",
"ui_fgcolor": "#000",
"ui_label": "end"
},
"section_1.task_1": {
"emits_ol_events": false,
"is_setup": false,
"is_teardown": false,
"operator": "airflow.operators.empty.EmptyOperator",
"task_group": "section_1",
"ui_color": "#e8f7e4",
"ui_fgcolor": "#000",
"ui_label": "task_1"
},
"section_1.task_2": {
"emits_ol_events": true,
"is_setup": false,
"is_teardown": false,
"operator": "airflow.operators.bash.BashOperator",
"task_group": "section_1",
"ui_color": "#f0ede4",
"ui_fgcolor": "#000",
"ui_label": "task_2"
},
"section_1.task_3": {
"emits_ol_events": false,
"is_setup": false,
"is_teardown": false,
"operator": "airflow.operators.empty.EmptyOperator",
"task_group": "section_1",
"ui_color": "#e8f7e4",
"ui_fgcolor": "#000",
"ui_label": "task_3"
},
"section_2.inner_section_2.task_2": {
"emits_ol_events": true,
"is_setup": false,
"is_teardown": false,
"operator": "airflow.operators.bash.BashOperator",
"task_group": "section_2.inner_section_2",
"ui_color": "#f0ede4",
"ui_fgcolor": "#000",
"ui_label": "task_2"
},
"section_2.inner_section_2.task_3": {
"emits_ol_events": false,
"is_setup": false,
"is_teardown": false,
"operator": "airflow.operators.empty.EmptyOperator",
"task_group": "section_2.inner_section_2",
"ui_color": "#e8f7e4",
"ui_fgcolor": "#000",
"ui_label": "task_3"
},
"section_2.inner_section_2.task_4": {
"emits_ol_events": false,
"is_setup": false,
"is_teardown": false,
"operator": "airflow.operators.empty.EmptyOperator",
"task_group": "section_2.inner_section_2",
"ui_color": "#e8f7e4",
"ui_fgcolor": "#000",
"ui_label": "task_4"
},
"section_2.task_1": {
"emits_ol_events": false,
"is_setup": false,
"is_teardown": false,
"operator": "airflow.operators.empty.EmptyOperator",
"task_group": "section_2",
"ui_color": "#e8f7e4",
"ui_fgcolor": "#000",
"ui_label": "task_1"
},
"start": {
"emits_ol_events": false,
"is_setup": false,
"is_teardown": false,
"operator": "airflow.operators.empty.EmptyOperator",
"ui_color": "#e8f7e4",
"ui_fgcolor": "#000",
"ui_label": "start"
}
}
},
"jobType": {
"_producer": "https://github.com/apache/airflow/tree/providers-openlineage/1.9.1",
"_schemaURL": "https://raw.githubusercontent.com/OpenLineage/OpenLineage/main/spec/OpenLineage.json#/definitions/JobTypeJobFacet",
"integration": "AIRFLOW",
"jobType": "DAG",
"processingType": "BATCH"
},
"ownership": {
"_producer": "https://github.com/apache/airflow/tree/providers-openlineage/1.9.1",
"_schemaURL": "https://raw.githubusercontent.com/OpenLineage/OpenLineage/main/spec/OpenLineage.json#/definitions/OwnershipJobFacet",
"owners": [
{
"name": "airflow"
}
]
}
},
"name": "example_task_group",
"namespace": "default"
},
"outputs": [],
"producer": "https://github.com/apache/airflow/tree/providers-openlineage/1.9.1",
"run": {
"facets": {
"airflow_dagRun": {
"_producer": "https://github.com/apache/airflow/tree/providers-openlineage/1.9.1",
"_schemaURL": "https://raw.githubusercontent.com/OpenLineage/OpenLineage/main/spec/OpenLineage.json#/definitions/BaseFacet",
"dag": {
"dag_id": "example_task_group",
"owner": "airflow",
"schedule_interval": "86400.0 seconds",
"start_date": "2021-01-01T00:00:00+00:00",
"tags": [
"example"
],
"timetable": {
"delta": 86400.0
}
},
"dagRun": {
"conf": {},
"dag_id": "example_task_group",
"data_interval_end": "2024-07-19T12:27:25.372580+00:00",
"data_interval_start": "2024-07-18T12:27:25.372580+00:00",
"external_trigger": true,
"run_id": "manual__2024-07-19T12:27:25.372580+00:00",
"run_type": "manual",
"start_date": "2024-07-19T12:27:26.050677+00:00"
}
},
"nominalTime": {
"_producer": "https://github.com/apache/airflow/tree/providers-openlineage/1.9.1",
"_schemaURL": "https://raw.githubusercontent.com/OpenLineage/OpenLineage/main/spec/OpenLineage.json#/definitions/NominalTimeRunFacet",
"nominalEndTime": "2024-07-19T12:27:25.372580+00:00",
"nominalStartTime": "2024-07-18T12:27:25.372580+00:00"
}
},
"runId": "0190caf6-1d3c-724f-856c-c0b8c295ec09"
},
"schemaURL": "https://openlineage.io/spec/1-0-5/OpenLineage.json#/definitions/RunEvent"
}
```
</details>
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on PR #40854:
URL: https://github.com/apache/airflow/pull/40854#issuecomment-2236823634
> > Why not attach it to some new facet as AirflowDagRunFacet?
>
> What should be the key of this facet? `"airflow"` or something else? For OL consumer, having runEvents with facets stored with the same key, but having different fields for Task and DAG runs, is not very comfortable.
>
> Should `AirflowRunFacet` be renamed to `AirflowTaskRunFacet`, for consistency?
We can think of a new name, that would not be confusing. I don't think we even could go with another run facet with `airflow` as key, as from spec perspective it would be a duplicate, there is no distinction between DAG and Task run facet from the spec perspective.
I think the renaming you propose would not be a big issue, as it would not change the spec, only the import, so we would have to keep old name for compatibility. Personally, I would avoid doing any changes to AirflowRunFacet if possible. I think renaming will not be necessary if we come up with a better name for the new one ๐
--
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] openlineage: Add AirflowDagRunFacet to dag runEvents [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski merged PR #40854:
URL: https://github.com/apache/airflow/pull/40854
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on PR #40854:
URL: https://github.com/apache/airflow/pull/40854#issuecomment-2236869935
> My current plan is:
>
> * Mark AirflowRunFacet as deprecated, but keep it for backward compatibility (al least for several next releases).
> * Add new facet `airflow_task: AirflowTaskRunFacet` which would just copy `AirflowRunFacet`, but with a new name and key.
> * Add new facet `airflow_dag: AirflowDagRunFacet` which would implement changes for [OpenLineage: Add AirflowRunFacet to DAG run eventsย #40798](https://github.com/apache/airflow/issues/40798).
>
> What do you think?
I think i would proceed with the last point for now in this PR, and create a discussion for the first two so that we can decide if / how we want to make it less confusing for users. The AirflowRunFacet is the primary facet that users rely on, so we need to handle it with care. Before proceeding with any actions, we should have an in-depth discussion, as deprecating it without such a discussion would not be advisable. ๐
We can also discuss here what would be the best key for this new facet: `airflowDag`, `airflowRun`, `airflowDagRun`?
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1684332607
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -307,6 +308,17 @@ class TaskGroupInfo(InfoJsonEncodable):
]
+def get_airflow_dag_run_facet(dag_run: DagRun) -> dict[str, BaseFacet]:
+ if not dag_run.dag:
+ return {}
+ return {
+ "airflow_dagRun": AirflowDagRunFacet(
Review Comment:
I don't have a strong opinion here, i think I simply want to avoid introducing multiple naming cases that the consumers would have to handle ๐
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1684242857
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -307,6 +308,17 @@ class TaskGroupInfo(InfoJsonEncodable):
]
+def get_airflow_dag_run_facet(dag_run: DagRun) -> dict[str, BaseFacet]:
+ if not dag_run.dag:
+ return {}
+ return {
+ "airflow_dagRun": AirflowDagRunFacet(
Review Comment:
Personally I'd stick with current name, even though it's not compliant with the doc. Maybe we should adjust it, but sending duplicates for a long time for newly created facet just because of this mistakes seems like on overkill for me. I think in more and more places we are going with camelCase instead (e.x. [BQ mixin](https://github.com/apache/airflow/blob/42a96412b4cbe309c5c9f0f298a5514de838e9d7/airflow/providers/google/cloud/openlineage/mixins.py#L127))
--
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] openlineage: Add AirflowDagRunFacet to dag runEvents [airflow]
Posted by "dolfinus (via GitHub)" <gi...@apache.org>.
dolfinus commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1686240389
##########
airflow/providers/openlineage/plugins/adapter.py:
##########
@@ -329,19 +330,26 @@ def dag_started(
self,
dag_run: DagRun,
msg: str,
- nominal_start_time: str,
- nominal_end_time: str,
job_facets: dict[str, BaseFacet] | None = None, # Custom job facets
+ run_facets: dict[str, BaseFacet] | None = None, # Custom run facets
):
+ if not run_facets:
+ run_facets = {}
Review Comment:
Removed
--
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] openlineage: Add AirflowDagRunFacet to dag runEvents [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1686369640
##########
airflow/providers/openlineage/plugins/adapter.py:
##########
@@ -329,19 +330,26 @@ def dag_started(
self,
dag_run: DagRun,
msg: str,
- nominal_start_time: str,
- nominal_end_time: str,
Review Comment:
Adapter methods are not public API.
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "dolfinus (via GitHub)" <gi...@apache.org>.
dolfinus commented on PR #40854:
URL: https://github.com/apache/airflow/pull/40854#issuecomment-2236511831
> Why not attach it to some new facet as AirflowDagRunFacet?
What should be the key of this facet? `"airflow"` or something else? For OL consumer, having runEvents with facets which looks the same and stored with the same key, but have different fields for Task and DAG runs, is not very comfortable.
Should `AirflowRunFacet` be renamed to `AirflowTaskRunFacet`, for consistency?
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1684207413
##########
airflow/providers/openlineage/plugins/listener.py:
##########
@@ -420,7 +421,8 @@ def on_dag_run_running(self, dag_run: DagRun, msg: str) -> None:
nominal_end_time=data_interval_end,
# AirflowJobFacet should be created outside ProcessPoolExecutor that pickles objects,
# as it causes lack of some TaskGroup attributes and crashes event emission.
- job_facets={**get_airflow_job_facet(dag_run=dag_run)},
+ job_facets=get_airflow_job_facet(dag_run=dag_run),
+ run_facets=get_airflow_dag_run_facet(dag_run),
Review Comment:
Try to test it manually as well and see if all needed attributes of dagrun and dag are accessible from the adapter. As mentioned in the comment in the code, i had some issue with accessing some task group details after pickling that happens when using ProcessPoolExecutor.
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1684242857
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -307,6 +308,17 @@ class TaskGroupInfo(InfoJsonEncodable):
]
+def get_airflow_dag_run_facet(dag_run: DagRun) -> dict[str, BaseFacet]:
+ if not dag_run.dag:
+ return {}
+ return {
+ "airflow_dagRun": AirflowDagRunFacet(
Review Comment:
Personally I'd stick with current name `airflowState`, even though it's not compliant with the doc. Maybe we should adjust the doc, but sending duplicates for a long time for newly created facet just because of this mistake and ole doc seems like on overkill for me. I think in more and more places we are going with camelCase instead (e.x. [BQ mixin](https://github.com/apache/airflow/blob/42a96412b4cbe309c5c9f0f298a5514de838e9d7/airflow/providers/google/cloud/openlineage/mixins.py#L127))
And for this facet, I'd go with `airflowDagRun`.
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -307,6 +308,17 @@ class TaskGroupInfo(InfoJsonEncodable):
]
+def get_airflow_dag_run_facet(dag_run: DagRun) -> dict[str, BaseFacet]:
+ if not dag_run.dag:
+ return {}
+ return {
+ "airflow_dagRun": AirflowDagRunFacet(
Review Comment:
Personally I'd stick with current name `airflowState`, even though it's not compliant with the doc. Maybe we should adjust the doc, but sending duplicates for a long time for newly created facet just because of this mistake and old doc seems like on overkill for me. I think in more and more places we are going with camelCase instead (e.x. [BQ mixin](https://github.com/apache/airflow/blob/42a96412b4cbe309c5c9f0f298a5514de838e9d7/airflow/providers/google/cloud/openlineage/mixins.py#L127))
And for this facet, I'd go with `airflowDagRun`.
--
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] openlineage: Add AirflowDagRunFacet to dag runEvents [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1686370968
##########
airflow/providers/openlineage/plugins/adapter.py:
##########
@@ -329,19 +330,26 @@ def dag_started(
self,
dag_run: DagRun,
msg: str,
- nominal_start_time: str,
- nominal_end_time: str,
Review Comment:
@dolfinus it's more likely we'll redesign the whole mechanism in Airflow 3 due to AIP-72 - and possibly stop running on worker altogether.
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "dolfinus (via GitHub)" <gi...@apache.org>.
dolfinus commented on PR #40854:
URL: https://github.com/apache/airflow/pull/40854#issuecomment-2236466647
> Added to AirflowJobFacet that we attach to DagRun events.
DagRun information is a part of RunFacet, and cannot be JobFacet.
> I think this should be a different facet
Is that okay if facet with key `"airflow"` will have different schema for DAG and Task?
Honestly, I'd rather split these fields to different facets, like:
```yaml
job:
facets:
airflow_dag: AirflowDagJobFacet(...) # previously airflow:dag
airflow_task: AirflowTaskJobFacet(...) # previously airflow:task
run:
facets:
airflow_dagRun: AirflowDagRunFacet(...) # previously airflow:dagRun
airflow_taskInstance: AirflowTaskInstanceFacet(...) # previously airflow:taskInstance
# throw away taskUuuid, it just ruplicates runId
```
But this is not backwards compatible.
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on PR #40854:
URL: https://github.com/apache/airflow/pull/40854#issuecomment-2236502492
Why not attach it to some new facet as `AirflowDagRunFacet`? Basically, `AirflowRunFacet` is tightly coupled to what we have during task instance execution, and don't think making it just a bag of possible things makes sense.
I'd rather have the same DagRun information attached to different facets in TaskInstance level events and DagRun events.
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "dolfinus (via GitHub)" <gi...@apache.org>.
dolfinus commented on PR #40854:
URL: https://github.com/apache/airflow/pull/40854#issuecomment-2236841435
My current plan is:
* Revert changes to AirflowRunFacet, mark it s deprecated, but keep it for backward compatibility (al least for several next releases).
* Add new facet `airflow_task: AirflowTaskRunFacet` which would just copy `AirflowRunFacet`, but with a new name and key.
* Add new facet `airflow_dag: AirflowDagRunFacet` which would implement changes for #40798.
What do you think?
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "dolfinus (via GitHub)" <gi...@apache.org>.
dolfinus commented on PR #40854:
URL: https://github.com/apache/airflow/pull/40854#issuecomment-2238815432
Updated implementation - current approach with `AirflowRunFacet` is left intact, new facet `airflow_dagRun: AirflowDagRunFacet` is introducet. Using key `airflow_dagRun` instead of `airflowDagRun` as recommended by [specification](https://github.com/OpenLineage/OpenLineage/blob/caecf5f39a4b2bb25513a3a0ef7a5346b4dd4a0d/spec/OpenLineage.md?plain=1#L120-L123).
Regarding changing `AirflowRunFacet` name, key and probably content I've created discussion https://github.com/apache/airflow/discussions/40886.
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "dolfinus (via GitHub)" <gi...@apache.org>.
dolfinus commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1684198103
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -307,6 +308,17 @@ class TaskGroupInfo(InfoJsonEncodable):
]
+def get_airflow_dag_run_facet(dag_run: DagRun) -> dict[str, BaseFacet]:
+ if not dag_run.dag:
+ return {}
+ return {
+ "airflow_dagRun": AirflowDagRunFacet(
Review Comment:
It is mentioned only once, in OpenLineage.md file, but in the docs on site
--
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] openlineage: Add AirflowRunFacet for dag runEvents [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #40854:
URL: https://github.com/apache/airflow/pull/40854#discussion_r1684242857
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -307,6 +308,17 @@ class TaskGroupInfo(InfoJsonEncodable):
]
+def get_airflow_dag_run_facet(dag_run: DagRun) -> dict[str, BaseFacet]:
+ if not dag_run.dag:
+ return {}
+ return {
+ "airflow_dagRun": AirflowDagRunFacet(
Review Comment:
Personally I'd stick with current name `airflowState`, even though it's not compliant with the doc. Maybe we should adjust the doc, but sending duplicates for a long time for newly created facet just because of this mistakes seems like on overkill for me. I think in more and more places we are going with camelCase instead (e.x. [BQ mixin](https://github.com/apache/airflow/blob/42a96412b4cbe309c5c9f0f298a5514de838e9d7/airflow/providers/google/cloud/openlineage/mixins.py#L127))
And for this facet, I'd go with `airflowDagRun`.
--
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