You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "pawelgrochowicz (via GitHub)" <gi...@apache.org> on 2025/09/12 14:21:10 UTC
[PR] feature: Add OpenLineage support for CloudDataFusionStartPipelineOperator [airflow]
pawelgrochowicz opened a new pull request, #55573:
URL: https://github.com/apache/airflow/pull/55573
This PR adds OpenLineage support for CloudDataFusionStartPipelineOperator.
<!-- 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 [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/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] feature: Add OpenLineage support for CloudDataFusionStartPipelineOperator [airflow]
Posted by "pawelgrochowicz (via GitHub)" <gi...@apache.org>.
pawelgrochowicz commented on code in PR #55573:
URL: https://github.com/apache/airflow/pull/55573#discussion_r2368566695
##########
providers/google/src/airflow/providers/google/cloud/operators/datafusion.py:
##########
@@ -863,6 +867,34 @@ def execute_complete(self, context: Context, event: dict[str, Any]):
)
return event["pipeline_id"]
+ def get_openlineage_facets_on_complete(self, task_instance) -> OperatorLineage | None:
+ """Build and return OpenLineage facets and datasets for the completed pipeline start."""
+ from airflow.providers.common.compat.openlineage.facet import Dataset
+ from airflow.providers.google.cloud.openlineage.facets import DataFusionRunFacet
+ from airflow.providers.openlineage.extractors import OperatorLineage
+
+ pipeline_resource = (
+ f"projects/{self.project_id}/locations/{self.location}/instances/"
+ f"{self.instance_name}/pipelines/{self.pipeline_name}"
+ )
+
+ inputs = [Dataset(namespace="datafusion", name=pipeline_resource)]
Review Comment:
I was following the Pub/Sub example from [the OpenLineage naming spec](https://openlineage.io/docs/spec/naming/#dataset-naming)
and the way the pattern is defined. In Pub/Sub, both the topic and the subscription are included in the name, for example:
topic:{projectId}:{topicId}
subscription:{projectId}:{subscriptionId}
The namespace in this case is simply the service name: "pubsub".
Another example is BigQuery, where the namespace is again just the service name, but the full details are included in the name.
For Data Fusion, things are slightly different. There is an instance, which acts as the container for all pipelines (similar to grouping multiple workflow definitions in one place). Within that instance, there are pipelines, which represent the actual workflows. This is why I was inspired by pubsub
--
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] feature: Add OpenLineage support for CloudDataFusionStartPipelineOperator [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on PR #55573:
URL: https://github.com/apache/airflow/pull/55573#issuecomment-3355276811
@pawelgrochowicz I think you need to rebase your PRs, there are some issues that look unrelated but fail the CI
--
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] feature: Add OpenLineage support for CloudDataFusionStartPipelineOperator [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #55573:
URL: https://github.com/apache/airflow/pull/55573#discussion_r2368376243
##########
providers/google/src/airflow/providers/google/cloud/operators/datafusion.py:
##########
@@ -863,6 +867,34 @@ def execute_complete(self, context: Context, event: dict[str, Any]):
)
return event["pipeline_id"]
+ def get_openlineage_facets_on_complete(self, task_instance) -> OperatorLineage | None:
+ """Build and return OpenLineage facets and datasets for the completed pipeline start."""
+ from airflow.providers.common.compat.openlineage.facet import Dataset
+ from airflow.providers.google.cloud.openlineage.facets import DataFusionRunFacet
+ from airflow.providers.openlineage.extractors import OperatorLineage
+
+ pipeline_resource = (
+ f"projects/{self.project_id}/locations/{self.location}/instances/"
+ f"{self.instance_name}/pipelines/{self.pipeline_name}"
+ )
+
+ inputs = [Dataset(namespace="datafusion", name=pipeline_resource)]
Review Comment:
Is this the right split between name and namespace? Not sure how Google data Fusion works underneath, but if it has some instances then maybe it should be something like:
name=`datafusion://projects/{self.project_id}/locations/{self.location}/instances/{self.instance_name}`
namespace=`pipelines/{self.pipeline_name}`
?
##########
providers/google/tests/unit/google/common/auth_backend/test_google_openid.py:
##########
@@ -23,6 +23,7 @@
# Do not run the tests when FAB / Flask is not installed
pytest.importorskip("flask_session")
+pytest.importorskip("flask_limiter")
Review Comment:
This seems unrelated to the PR, why is that added?
--
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] feature: Add OpenLineage support for CloudDataFusionStartPipelineOperator [airflow]
Posted by "pawelgrochowicz (via GitHub)" <gi...@apache.org>.
pawelgrochowicz commented on code in PR #55573:
URL: https://github.com/apache/airflow/pull/55573#discussion_r2368409025
##########
providers/google/tests/unit/google/common/auth_backend/test_google_openid.py:
##########
@@ -23,6 +23,7 @@
# Do not run the tests when FAB / Flask is not installed
pytest.importorskip("flask_session")
+pytest.importorskip("flask_limiter")
Review Comment:
Something has changed in the dependencies of the fab provider, so Google provider tests failed (Python version compatibility issue). I just ignored the missing dependency in tests as the community noticed this issue but planned to fix it later than I created my PR.
--
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] feature: Add OpenLineage support for CloudDataFusionStartPipelineOperator [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on PR #55573:
URL: https://github.com/apache/airflow/pull/55573#issuecomment-3355830870
rebased. lets see
--
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] feature: Add OpenLineage support for CloudDataFusionStartPipelineOperator [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on PR #55573:
URL: https://github.com/apache/airflow/pull/55573#issuecomment-3365713530
@pawelgrochowicz I think some git error happened here, there are unrelated commits included in this PR
--
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] feature: Add OpenLineage support for CloudDataFusionStartPipelineOperator [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on PR #55573:
URL: https://github.com/apache/airflow/pull/55573#issuecomment-3365354652
@pawelgrochowicz merged previous PR and now it has conflicts to fix :)
--
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] feature: Add OpenLineage support for CloudDataFusionStartPipelineOperator [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #55573:
URL: https://github.com/apache/airflow/pull/55573#discussion_r2368389507
##########
providers/google/src/airflow/providers/google/cloud/operators/datafusion.py:
##########
@@ -863,6 +867,34 @@ def execute_complete(self, context: Context, event: dict[str, Any]):
)
return event["pipeline_id"]
+ def get_openlineage_facets_on_complete(self, task_instance) -> OperatorLineage | None:
+ """Build and return OpenLineage facets and datasets for the completed pipeline start."""
+ from airflow.providers.common.compat.openlineage.facet import Dataset
+ from airflow.providers.google.cloud.openlineage.facets import DataFusionRunFacet
+ from airflow.providers.openlineage.extractors import OperatorLineage
+
+ pipeline_resource = (
+ f"projects/{self.project_id}/locations/{self.location}/instances/"
+ f"{self.instance_name}/pipelines/{self.pipeline_name}"
+ )
+
+ inputs = [Dataset(namespace="datafusion", name=pipeline_resource)]
Review Comment:
Let's also modify [OL naming convention](https://openlineage.io/docs/spec/naming) with whatever we choose
--
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] feature: Add OpenLineage support for CloudDataFusionStartPipelineOperator [airflow]
Posted by "pawelgrochowicz (via GitHub)" <gi...@apache.org>.
pawelgrochowicz closed pull request #55573: feature: Add OpenLineage support for CloudDataFusionStartPipelineOperator
URL: https://github.com/apache/airflow/pull/55573
--
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] feature: Add OpenLineage support for CloudDataFusionStartPipelineOperator [airflow]
Posted by "pawelgrochowicz (via GitHub)" <gi...@apache.org>.
pawelgrochowicz commented on PR #55573:
URL: https://github.com/apache/airflow/pull/55573#issuecomment-3364506660
@mobuchowski, rebased and now checks are good
--
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] feature: Add OpenLineage support for CloudDataFusionStartPipelineOperator [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #55573:
URL: https://github.com/apache/airflow/pull/55573#issuecomment-3286258684
Compat issue, but generally looks fine.
--
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] feature: Add OpenLineage support for CloudDataFusionStartPipelineOperator [airflow]
Posted by "pawelgrochowicz (via GitHub)" <gi...@apache.org>.
pawelgrochowicz commented on PR #55573:
URL: https://github.com/apache/airflow/pull/55573#issuecomment-3365893924
As this PR has been too messy, I closed this one and created a new one with just my code https://github.com/apache/airflow/pull/56365
--
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