kacpermuda opened a new pull request, #65788:
URL: https://github.com/apache/airflow/pull/65788
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull
request.
Write a good git commit message following this guide:
http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the
documentation.
Feel free to ping (in general) for the review if you do not see reaction for
a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we
sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
## Why
The OpenLineage provider is unlike most Airflow providers: it does not ship
user-facing operators or hooks. Instead, it runs in the background via a
listener and is consumed by *other* providers. Until now, DAG authors who
wanted to emit their own lineage from a `@task` or `PythonOperator` had to
reach into `airflow.providers.openlineage.plugins.*` internals - modules with
no stability guarantees, prone to breaking between releases.
This PR introduces a dedicated `airflow.providers.openlineage.api` package
as the **single public, stable entry point** for DAG authors and third-party
providers.
## What's new
Two user-facing emitters:
- **`emit_dataset_lineage(inputs=..., outputs=..., ...)`** - send
input/output `Dataset`s from currently running task. Emits a RUNNING event that
peers with the listener's START/COMPLETE events (same run id, parent,
ownership, tags, nominalTime - so it shows up correctly for OpenLineage
consumers).
- **`emit_query_lineage(query_text=..., query_source_namespace=...,
query_id=..., ...)`** - emit a START + COMPLETE/FAIL event pair describing a
single SQL query executed by the task. When `query_text` and
`query_source_namespace` are provided, inputs/outputs are parsed directly from
the SQL via the provider's SQL parser; when `query_id` is provided, an
`externalQuery` facet links the event to the warehouse-side query id.
Both resolve the current task instance automatically from the execution
context, so the common case is a one-liner. `emit_query_lineage` can also be
called multiple times within a single task - each call produces a distinct,
sequentially-numbered job.
## Example
```python
from airflow.decorators import task
from openlineage.client.event_v2 import Dataset
from airflow.providers.openlineage.api import emit_dataset_lineage,
emit_query_lineage
@task
def ingest():
# ... call an API, write a file, etc. ...
emit_dataset_lineage(
inputs=[Dataset(namespace="s3://bucket", name="raw/2024/01/01.csv")],
outputs=[Dataset(namespace="snowflake://acct",
name="analytics.public.users")],
)
@task
def transform():
sql = "INSERT INTO analytics.summary SELECT id, count(*) FROM
analytics.users GROUP BY id"
# ... run the query via your client of choice, capture its id ...
emit_query_lineage(
query_id="acde070d-8c4c-4f0d-9d8a-162843c10333",
query_source_namespace="snowflake://acct",
query_text=sql,
)
```
Typical use cases: `PythonOperator` / `@task` functions that call APIs or
run SQL directly and third-party providers that want to emit lineage without
depending on provider internals.
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR,
please
change below checkbox to `[X]` followed by the name of the tool, uncomment
the "Generated-by".
-->
- [X] Yes (please specify the tool below)
Generated-by: Claude following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
---
* Read the **[Pull Request
Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#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](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals))
is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License
Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment:
`{pr_number}.significant.rst`, in
[airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
You can add this file in a follow-up commit after the PR is created so you
know the PR number.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
