New google dataflow operator to fetch job metrics

[pushpendu1991]

A new Google dataflow operator that fetches metrics from Google Cloud Dataflow jobs and routes them to one or more destinations (Pub/Sub, BigQuery, or both) in a single operation. The operator supports both synchronous blocking and asynchronous deferrable execution modes for optimal resource utilization.

Apache Airflow's Google Cloud provider package does not include a native operator for:

1. Programmatic Dataflow Job Metrics Collection — Users currently rely on dataflow hook or dataflow API to collect Dataflow job metrics. There is no built-in Airflow operator to fetch metrics directly from a Dataflow job within a DAG.

2. Multi-Destination Metrics Routing — Enterprises often need to send the same metrics to multiple systems for different purposes (e.g., Pub/Sub for real-time alerting, BigQuery for historical analysis). There is no simple way to fan-out metrics to multiple destinations atomically.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[Copilot]

Pull request overview

Note

Copilot was unable to run its full agentic suite in this review.

Adds a new Google Cloud Dataflow operator to fetch job metrics and optionally route them to Pub/Sub and/or BigQuery, supporting both synchronous and deferrable execution, along with unit/system tests and documentation.

Changes:

• Introduces DataflowGetMetricsOperator with routing to Pub/Sub and BigQuery, including deferrable trigger-based execution.
• Adds unit tests covering init, sync routing scenarios, deferrable execution, and execute_complete.
• Adds a system test example DAG and updates Dataflow operator docs with usage snippets.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 10 comments.

File	Description
providers/google/src/airflow/providers/google/cloud/operators/dataflow.py	Implements DataflowGetMetricsOperator including routing and deferrable support.
providers/google/tests/unit/google/cloud/operators/test_dataflow.py	Adds unit tests for the new operator’s behavior.
providers/google/tests/system/google/cloud/dataflow/example_dataflow_get_metrics.py	Adds a system-test example DAG demonstrating operator usage.
providers/google/docs/operators/cloud/dataflow.rst	Documents the new operator with example includes.