You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "anandhimurali (via GitHub)" <gi...@apache.org> on 2024/04/12 23:33:21 UTC
[PR] Ability to add custom facet in OpenLineage events [airflow]
anandhimurali opened a new pull request, #38982:
URL: https://github.com/apache/airflow/pull/38982
**Motivation**
Currently, users need to write custom extractor inorder to add custom facets in the lineage events. Managing the entire extractor will be hard while users might just want to add few custom facets as per their needs.
**Modifications**
The `get_custom_facets` function in the OpenLineage provider is extended to fetch the Airflow config `AIRFLOW__OPENLINEAGE__CUSTOM_FACET_FUNCTIONS`, execute those functions and append their return values to the run facet of the lineage event generated by the Airflow extractor.
**Result**
This PR enables users to add custom factes to the lineage events by defining functions that return the custom facet and setting the fully qualified path to the functions as an Airflow config `AIRFLOW__OPENLINEAGE__CUSTOM_FACET_FUNCTIONS`. After this change, users need not write the complete extractor in order to add custom facet in the lineage event.
**Configs documentation for `AIRFLOW__OPENLINEAGE__CUSTOM_FACET_FUNCTIONS`**
<!--
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.
-->
<!--
Thank you for contributing! 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 committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
<!-- 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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #38982:
URL: https://github.com/apache/airflow/pull/38982#issuecomment-2052695209
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 [pre-commits]( https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst#prerequisites-for-pre-commit-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](https://github.com/apache/airflow/blob/main/docs/apache-airflow/howto/custom-operator.rst) Consider adding an example DAG that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/dev/breeze/doc/README.rst) 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](https://www.apache.org/foundation/policies/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]( https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#coding-style-and-best-practices).
- 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
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1602585312
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -61,11 +61,20 @@ def get_job_name(task: TaskInstance) -> str:
def get_custom_facets(task_instance: TaskInstance | None = None) -> dict[str, Any]:
+ from airflow.providers.openlineage.extractors.manager import try_import_from_string
+
custom_facets = {}
# check for -1 comes from SmartSensor compatibility with dynamic task mapping
# this comes from Airflow code
if hasattr(task_instance, "map_index") and getattr(task_instance, "map_index") != -1:
custom_facets["airflow_mappedTask"] = AirflowMappedTaskRunFacet.from_task_instance(task_instance)
+
+ # Append custom run facets by executing the custom_facet_functions.
+ for custom_facet_func in conf.custom_facet_functions():
+ func: type[function] = try_import_from_string(custom_facet_func)
+ facet = func(task_instance) if func else None
+ if facet and isinstance(facet, dict):
+ custom_facets.update(facet)
Review Comment:
At the [place where the run facets are merged](https://github.com/apache/airflow/blob/9ea78d9d726d9ddb6109a7fba0c3a838f8a05610/airflow/providers/openlineage/plugins/listener.py#L147), the results from `get_custom_facets` is unwrapped before the `get_airflow_run_facet`. So user can't overwrite the Airflow base facets.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1665780045
##########
airflow/providers/openlineage/provider.yaml:
##########
@@ -104,6 +104,13 @@ config:
example: full.path.to.ExtractorClass;full.path.to.AnotherExtractorClass
default: ~
version_added: ~
+ custom_run_facets:
+ description: |
+ Register custom run facet functions by passing a string of semicolon separated full import paths.
+ type: string
+ example: full.path.to.custom_facet_function;full.path.to.another_custom_facet_function
+ default: ''
+ version_added: 1.8.0
Review Comment:
This will probably be 1.10.0 as 1.9.0 should be released before this gets merged.
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -78,11 +78,38 @@ def get_job_name(task: TaskInstance) -> str:
def get_custom_facets(task_instance: TaskInstance | None = None) -> dict[str, Any]:
+ from airflow.providers.openlineage.extractors.manager import try_import_from_string
+
custom_facets = {}
# check for -1 comes from SmartSensor compatibility with dynamic task mapping
# this comes from Airflow code
if hasattr(task_instance, "map_index") and getattr(task_instance, "map_index") != -1:
custom_facets["airflow_mappedTask"] = AirflowMappedTaskRunFacet.from_task_instance(task_instance)
+
+ # Append custom run facets by executing the custom_run_facets.
Review Comment:
It's more general comment but currently `get_custom_facets()` function is called only once, after task instance has started.
I believe we need to decide whether custom run facets should receive the task before or after execution. Users might want to use elements like `query_id` generated during execution in their custom facets. While I don't have a strong preference either way, it's important to note this decision consciously in the documentation.
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -78,11 +78,38 @@ def get_job_name(task: TaskInstance) -> str:
def get_custom_facets(task_instance: TaskInstance | None = None) -> dict[str, Any]:
+ from airflow.providers.openlineage.extractors.manager import try_import_from_string
+
custom_facets = {}
# check for -1 comes from SmartSensor compatibility with dynamic task mapping
# this comes from Airflow code
if hasattr(task_instance, "map_index") and getattr(task_instance, "map_index") != -1:
custom_facets["airflow_mappedTask"] = AirflowMappedTaskRunFacet.from_task_instance(task_instance)
+
+ # Append custom run facets by executing the custom_run_facets.
+ for custom_facet_func in conf.custom_run_facets():
+ func: type[function] | None = try_import_from_string(custom_facet_func)
+ if not func:
+ log.warning(
+ "OpenLineage is unable to import custom facet function `%s`; will ignore it.",
+ custom_facet_func,
+ )
+ continue
+ facet: dict[str, BaseFacet] = func(task_instance)
Review Comment:
And one more question / decision to make: should we wrap the function call with try/except to avoid crashing the whole event emission or do we let the user crash it if they provided the function that breaks? I would vote for putting a try/except with some descriptive logging here.
##########
docs/apache-airflow-providers-openlineage/guides/developer.rst:
##########
@@ -446,15 +446,96 @@ Conversion from Airflow Table entity to OpenLineage Dataset is made in the follo
.. _custom_facets:openlineage:
-Custom facets
+Custom Facets
=============
To learn more about facets in OpenLineage, please refer to `facet documentation <https://openlineage.io/docs/spec/facets/>`_.
-Also check out `available Facets <https://github.com/OpenLineage/OpenLineage/blob/main/client/python/openlineage/client/facet.py>`_
+Also check out `available facets <https://github.com/OpenLineage/OpenLineage/blob/main/client/python/openlineage/client/facet.py>`_
The OpenLineage spec might not contain all the facets you need to write your extractor,
in which case you will have to make your own `custom facets <https://openlineage.io/docs/spec/facets/custom-facets>`_.
More on creating custom facets can be found `here <https://openlineage.io/blog/extending-with-facets/>`_.
+Custom Run Facets
+-----------------
+
+You can inject your own custom facets in the lineage event's run facet using the ``custom_run_facets`` Airflow configuration.
+
+Steps to be taken,
+
+1. Write a function that returns the custom facet. You can write as many custom facet functions as needed.
+2. Register the functions using the ``custom_run_facets`` Airflow configuration.
+
+Once done, Airflow OpenLineage listener will automatically execute these functions during the lineage event generation
+and append their return values to the run facet in the lineage event.
+
+Writing a custom facet function
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+- **Input arguments:** The function should accept the ``TaskInstance`` as an input argument.
+- **Function body:** Perform the logic needed to generate the custom facet. The custom facet should inherit from the ``BaseFacet`` for the ``_producer`` and ``_schemaURL`` to be automatically added for the facet.
+- **Return value:** The custom facet to be added to the lineage event. Return type should be ``None`` or ``dict``. You may choose to return ``None`` or ``{}``, if you do not want to add custom facets for certain criteria.
+
+**Example custom facet function**
+
+.. code-block:: python
+
+ import attrs
+ from airflow.models import TaskInstance
+ from openlineage.client.facet import BaseFacet
+
+
+ @attrs.define(slots=False)
+ class MyCustomRunFacet(BaseFacet):
+ """Define a custom facet."""
+
+ name: str
+ jobState: str
+ uniqueName: str
+ displayName: str
+ dagId: str
+ taskId: str
+ cluster: str
+
+
+ def get_my_custom_facet(task_instance: TaskInstance):
+ job_unique_name = f"TEST.{task_instance.dag_id}.{task_instance.task_id}"
Review Comment:
```suggestion
def get_my_custom_facet(task_instance: TaskInstance):
operator_name = task_instance.task.operator_name
if operator_name == "BashOperator":
return None
job_unique_name = f"TEST.{task_instance.dag_id}.{task_instance.task_id}"
```
Including some simple example here can make it easier for the users to add conditional facets.
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -78,11 +78,38 @@ def get_job_name(task: TaskInstance) -> str:
def get_custom_facets(task_instance: TaskInstance | None = None) -> dict[str, Any]:
+ from airflow.providers.openlineage.extractors.manager import try_import_from_string
+
custom_facets = {}
# check for -1 comes from SmartSensor compatibility with dynamic task mapping
# this comes from Airflow code
if hasattr(task_instance, "map_index") and getattr(task_instance, "map_index") != -1:
custom_facets["airflow_mappedTask"] = AirflowMappedTaskRunFacet.from_task_instance(task_instance)
+
+ # Append custom run facets by executing the custom_run_facets.
+ for custom_facet_func in conf.custom_run_facets():
+ func: type[function] | None = try_import_from_string(custom_facet_func)
+ if not func:
+ log.warning(
+ "OpenLineage is unable to import custom facet function `%s`; will ignore it.",
+ custom_facet_func,
+ )
+ continue
+ facet: dict[str, BaseFacet] = func(task_instance)
Review Comment:
I noticed that in examples here we are using attrs.asdict(), so we are providing the facets as dicts, but here we actually expect type[BaseFacet]. Could you adjust the typing and examples so that it matches?
##########
docs/apache-airflow-providers-openlineage/guides/developer.rst:
##########
@@ -446,15 +446,96 @@ Conversion from Airflow Table entity to OpenLineage Dataset is made in the follo
.. _custom_facets:openlineage:
-Custom facets
+Custom Facets
=============
To learn more about facets in OpenLineage, please refer to `facet documentation <https://openlineage.io/docs/spec/facets/>`_.
-Also check out `available Facets <https://github.com/OpenLineage/OpenLineage/blob/main/client/python/openlineage/client/facet.py>`_
+Also check out `available facets <https://github.com/OpenLineage/OpenLineage/blob/main/client/python/openlineage/client/facet.py>`_
The OpenLineage spec might not contain all the facets you need to write your extractor,
in which case you will have to make your own `custom facets <https://openlineage.io/docs/spec/facets/custom-facets>`_.
More on creating custom facets can be found `here <https://openlineage.io/blog/extending-with-facets/>`_.
+Custom Run Facets
+-----------------
+
+You can inject your own custom facets in the lineage event's run facet using the ``custom_run_facets`` Airflow configuration.
+
+Steps to be taken,
+
+1. Write a function that returns the custom facet. You can write as many custom facet functions as needed.
+2. Register the functions using the ``custom_run_facets`` Airflow configuration.
+
+Once done, Airflow OpenLineage listener will automatically execute these functions during the lineage event generation
+and append their return values to the run facet in the lineage event.
+
+Writing a custom facet function
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+- **Input arguments:** The function should accept the ``TaskInstance`` as an input argument.
+- **Function body:** Perform the logic needed to generate the custom facet. The custom facet should inherit from the ``BaseFacet`` for the ``_producer`` and ``_schemaURL`` to be automatically added for the facet.
+- **Return value:** The custom facet to be added to the lineage event. Return type should be ``None`` or ``dict``. You may choose to return ``None`` or ``{}``, if you do not want to add custom facets for certain criteria.
+
+**Example custom facet function**
+
+.. code-block:: python
+
+ import attrs
+ from airflow.models import TaskInstance
+ from openlineage.client.facet import BaseFacet
+
+
+ @attrs.define(slots=False)
+ class MyCustomRunFacet(BaseFacet):
+ """Define a custom facet."""
+
+ name: str
+ jobState: str
+ uniqueName: str
+ displayName: str
+ dagId: str
+ taskId: str
+ cluster: str
+
+
+ def get_my_custom_facet(task_instance: TaskInstance):
+ job_unique_name = f"TEST.{task_instance.dag_id}.{task_instance.task_id}"
Review Comment:
If we decide that this function will receive the task after the execution, we can also show in the example how to retrieve some attribute from the operator.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "ahidalgob (via GitHub)" <gi...@apache.org>.
ahidalgob commented on PR #38982:
URL: https://github.com/apache/airflow/pull/38982#issuecomment-2205728582
I think it would be good to rename this functionality to `custom_run_facets`, I think we might need to inject other facets too, specifically `Job.facets`.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "MassyB (via GitHub)" <gi...@apache.org>.
MassyB commented on PR #38982:
URL: https://github.com/apache/airflow/pull/38982#issuecomment-2402792332
@mobuchowski do you think this is only useful for Airflow ? I would like to have the same configuration for `dbt-ol` as well.
The signature of the function would be different (an empty args function is enough for my usecase)
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on PR #38982:
URL: https://github.com/apache/airflow/pull/38982#issuecomment-2076821285
Hey @anandhimurali @XD-DENG I think ability to add custom facets to arbitrary operators is a great idea.
I have two points I think it's worth addressing:
1. Whether we pass all the necessary info. I think given task_instance we can call `get_template_context` and get all the data, so this might not be an issue.
2. Whether this acts "per DAG/Task" or it's a mechanism that works for all the tasks equally - this might be defined inside the function so it still might not be an issue other than complicating the function a bit more.
Overall I like the solution, but might be good to specify the goals regarding 1) and 2) - because for sure, later, someone will come and ask "how to make this custom facet work only on some particular tasks" and "how do I pass some custom info to my function".
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on PR #38982:
URL: https://github.com/apache/airflow/pull/38982#issuecomment-2200253493
@anandhimurali any interest in getting this PR into shape and approved? I think this is good, really useful feature.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #38982:
URL: https://github.com/apache/airflow/pull/38982#issuecomment-2208558429
> Getting some documentation error on `breeze build-docs openlineage`. Trying to debug and fix this.
>
> <img alt="image" width="988" src="https://private-user-images.githubusercontent.com/68486238/345697063-3a27cddb-334e-441f-a63e-da7ba9eb4e35.png?jwt=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3MjAwODY1MTAsIm5iZiI6MTcyMDA4NjIxMCwicGF0aCI6Ii82ODQ4NjIzOC8zNDU2OTcwNjMtM2EyN2NkZGItMzM0ZS00NDFmLWE2M2UtZGE3YmE5ZWI0ZTM1LnBuZz9YLUFtei1BbGdvcml0aG09QVdTNC1ITUFDLVNIQTI1NiZYLUFtei1DcmVkZW50aWFsPUFLSUFWQ09EWUxTQTUzUFFLNFpBJTJGMjAyNDA3MDQlMkZ1cy1lYXN0LTElMkZzMyUyRmF3czRfcmVxdWVzdCZYLUFtei1EYXRlPTIwMjQwNzA0VDA5NDMzMFomWC1BbXotRXhwaXJlcz0zMDAmWC1BbXotU2lnbmF0dXJlPTIzZjJiZGYxYWRhMjY0YTNhOTMzMDdjZmZhZjk2YzJmNTU0NmRlOWJkNDk0ZTc0MGJlOTRhYTc3OTEwZDA1ZjQmWC1BbXotU2lnbmVkSGVhZGVycz1ob3N0JmFjdG9yX2lkPTAma2V5X2lkPTAmcmVwb19pZD0wIn0.b4yfnF-cygD1IpAQjJ4PXV89DEMcPf4N1d2Yf2XE2Ng">
FYI. Sometimes `--clean-build` flag helps.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski merged PR #38982:
URL: https://github.com/apache/airflow/pull/38982
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1684583435
##########
tests/system/providers/google/cloud/datapipelines/resources/__init__.py:
##########
Review Comment:
This was added by the pre-commit hook. I tried again now by removing the file and able to get pre-commit checks pass. So removed this file in 18526d7aa317666825d2c737f41d353ef8f5f62e.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1602580045
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -61,11 +61,20 @@ def get_job_name(task: TaskInstance) -> str:
def get_custom_facets(task_instance: TaskInstance | None = None) -> dict[str, Any]:
+ from airflow.providers.openlineage.extractors.manager import try_import_from_string
+
custom_facets = {}
# check for -1 comes from SmartSensor compatibility with dynamic task mapping
# this comes from Airflow code
if hasattr(task_instance, "map_index") and getattr(task_instance, "map_index") != -1:
custom_facets["airflow_mappedTask"] = AirflowMappedTaskRunFacet.from_task_instance(task_instance)
+
+ # Append custom run facets by executing the custom_facet_functions.
+ for custom_facet_func in conf.custom_facet_functions():
+ func: type[function] = try_import_from_string(custom_facet_func)
Review Comment:
Addressed in a3014683241b130ad9729dc462efede1160361e4
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1564573821
##########
airflow/providers/openlineage/provider.yaml:
##########
@@ -101,6 +101,15 @@ config:
example: full.path.to.ExtractorClass;full.path.to.AnotherExtractorClass
default: ~
version_added: ~
+ custom_facet_functions:
+ description: |
+ Register functions that return custom facets to append to the run facet. This config takes a
+ string of semicolon delimited full import paths to the functions. The functions should accept
+ a TaskInstance input argument and should return a dictionary.
+ type: string
+ example: full.path.to.custom_facet_function;full.path.to.another_custom_facet_function
+ default: ''
+ version_added: ~
Review Comment:
```suggestion
version_added: 1.8.0
```
We can assume this will be added in next feature release
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1607990291
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -61,11 +61,31 @@ def get_job_name(task: TaskInstance) -> str:
def get_custom_facets(task_instance: TaskInstance | None = None) -> dict[str, Any]:
+ from airflow.providers.openlineage.extractors.manager import try_import_from_string
+
custom_facets = {}
# check for -1 comes from SmartSensor compatibility with dynamic task mapping
# this comes from Airflow code
if hasattr(task_instance, "map_index") and getattr(task_instance, "map_index") != -1:
custom_facets["airflow_mappedTask"] = AirflowMappedTaskRunFacet.from_task_instance(task_instance)
+
+ # Append custom run facets by executing the custom_facet_functions.
+ for custom_facet_func in conf.custom_facet_functions():
+ func: type[function] | None = try_import_from_string(custom_facet_func)
+ if not func:
+ log.warning("Unable to import `%s`, will ignore it.", custom_facet_func)
+ facet = func(task_instance) if func else None
+ if facet and isinstance(facet, dict):
+ duplicate_facet_keys = [facet_key for facet_key in facet.keys() if facet_key in custom_facets]
+ if duplicate_facet_keys:
+ log.warning(
+ "Got duplicate facets key(s), `%s` from `%s`, will ignore it.",
+ ", ".join(duplicate_facet_keys),
+ custom_facet_func,
+ )
+ else:
+ log.info(f"Appending custom facets from {custom_facet_func}.")
+ custom_facets.update(facet)
Review Comment:
```suggestion
duplicate_facet_keys = [facet_key for facet_key in facet.keys() if facet_key in custom_facets]
if duplicate_facet_keys:
log.warning(
"Duplicate OpenLineage custom facets key(s) found: `%s` from function `%s`.",
", ".join(duplicate_facet_keys),
custom_facet_func,
)
log.debug("Adding OpenLineage custom facet with key(s): `%s` from function `%s`.", tuple(facet), custom_facet_func)
custom_facets.update(facet)
```
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -61,11 +61,31 @@ def get_job_name(task: TaskInstance) -> str:
def get_custom_facets(task_instance: TaskInstance | None = None) -> dict[str, Any]:
+ from airflow.providers.openlineage.extractors.manager import try_import_from_string
+
custom_facets = {}
# check for -1 comes from SmartSensor compatibility with dynamic task mapping
# this comes from Airflow code
if hasattr(task_instance, "map_index") and getattr(task_instance, "map_index") != -1:
custom_facets["airflow_mappedTask"] = AirflowMappedTaskRunFacet.from_task_instance(task_instance)
+
+ # Append custom run facets by executing the custom_facet_functions.
+ for custom_facet_func in conf.custom_facet_functions():
+ func: type[function] | None = try_import_from_string(custom_facet_func)
+ if not func:
+ log.warning("Unable to import `%s`, will ignore it.", custom_facet_func)
+ facet = func(task_instance) if func else None
Review Comment:
```suggestion
if not func:
log.warning("OpenLineage is unable to import custom facet function `%s`; will ignore it.", custom_facet_func)
continue
facets: dict[str, BaseFacet] = func(task_instance)
```
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on PR #38982:
URL: https://github.com/apache/airflow/pull/38982#issuecomment-2234939275
>@anandhimurali If you need any help to make the CI pass let me know, If you run the pre-commits locally then all of the static checks will pass. When it comes to 2.7 compatibility tests, you simply need to add start_date to the test DAGs you are creating and it should also work fine.
@kacpermuda, Thanks for letting me know about this. The necessary fixes are addressed in 81a8282b60c5bb28e46b39672ca99707dd7c7454 after the pre-commit checks.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1666331170
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -78,11 +78,38 @@ def get_job_name(task: TaskInstance) -> str:
def get_custom_facets(task_instance: TaskInstance | None = None) -> dict[str, Any]:
+ from airflow.providers.openlineage.extractors.manager import try_import_from_string
+
custom_facets = {}
# check for -1 comes from SmartSensor compatibility with dynamic task mapping
# this comes from Airflow code
if hasattr(task_instance, "map_index") and getattr(task_instance, "map_index") != -1:
custom_facets["airflow_mappedTask"] = AirflowMappedTaskRunFacet.from_task_instance(task_instance)
+
+ # Append custom run facets by executing the custom_run_facets.
Review Comment:
>It's more general comment but currently get_custom_facets() function is called only once, after task instance has started.
Updated documentation to capture this, 59bf11c56a5e8d3075894cd8a166717b82a4bc80.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1602602899
##########
airflow/providers/openlineage/conf.py:
##########
@@ -65,6 +65,19 @@ def custom_extractors() -> set[str]:
return set(extractor.strip() for extractor in option.split(";") if extractor.strip())
+@cache
+def custom_facet_functions() -> set[str]:
+ """[openlineage] custom_facet_functions."""
+ option = conf.get(_CONFIG_SECTION, "custom_facet_functions", fallback="")
+ if not option:
+ option = os.getenv("OPENLINEAGE_CUSTOM_FACET_FUNCTIONS", "")
Review Comment:
>I think good question is whether we can do it without global configs? I think this might be good to be configurable on task/dag/global level, rather than fixed for whole Airflow instance. It goes well with our effort to have lineage be incrementally enabled.
User can choose to `return None` in the custom facet function, if they don't want to add the custom facet for certain tasks/dags. Thinking if the complexity is going to increase if we handle the task/dag/global level settings at config level.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1602593146
##########
airflow/providers/openlineage/conf.py:
##########
@@ -65,6 +65,19 @@ def custom_extractors() -> set[str]:
return set(extractor.strip() for extractor in option.split(";") if extractor.strip())
+@cache
+def custom_facet_functions() -> set[str]:
+ """[openlineage] custom_facet_functions."""
+ option = conf.get(_CONFIG_SECTION, "custom_facet_functions", fallback="")
+ if not option:
+ option = os.getenv("OPENLINEAGE_CUSTOM_FACET_FUNCTIONS", "")
Review Comment:
>I don't think the OPENLINEAGE_CUSTOM_FACET_FUNCTIONS variable is necessary, we can stick to Airflow config. We use OPENLINEAGE_ variables in some other configs as they are legacy solutions that still need to be supported. For new features we can omit them.
Addressed in ec330478bbefc00118effe6ec9f4930907e3d273
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1684148107
##########
tests/system/providers/google/cloud/datapipelines/resources/__init__.py:
##########
Review Comment:
This file looks like some leftover from another branch
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1667195011
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -78,11 +78,38 @@ def get_job_name(task: TaskInstance) -> str:
def get_custom_facets(task_instance: TaskInstance | None = None) -> dict[str, Any]:
+ from airflow.providers.openlineage.extractors.manager import try_import_from_string
+
custom_facets = {}
# check for -1 comes from SmartSensor compatibility with dynamic task mapping
# this comes from Airflow code
if hasattr(task_instance, "map_index") and getattr(task_instance, "map_index") != -1:
custom_facets["airflow_mappedTask"] = AirflowMappedTaskRunFacet.from_task_instance(task_instance)
+
+ # Append custom run facets by executing the custom_run_facets.
+ for custom_facet_func in conf.custom_run_facets():
+ func: type[function] | None = try_import_from_string(custom_facet_func)
+ if not func:
+ log.warning(
+ "OpenLineage is unable to import custom facet function `%s`; will ignore it.",
+ custom_facet_func,
+ )
+ continue
+ facet: dict[str, BaseFacet] = func(task_instance)
Review Comment:
Return type can be `facet: dict[str, dict] | None`. Addressed in b5d3f37b5aa4171d112823ade8c41bf12743ef8b.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1666300729
##########
docs/apache-airflow-providers-openlineage/guides/developer.rst:
##########
@@ -446,15 +446,96 @@ Conversion from Airflow Table entity to OpenLineage Dataset is made in the follo
.. _custom_facets:openlineage:
-Custom facets
+Custom Facets
=============
To learn more about facets in OpenLineage, please refer to `facet documentation <https://openlineage.io/docs/spec/facets/>`_.
-Also check out `available Facets <https://github.com/OpenLineage/OpenLineage/blob/main/client/python/openlineage/client/facet.py>`_
+Also check out `available facets <https://github.com/OpenLineage/OpenLineage/blob/main/client/python/openlineage/client/facet.py>`_
The OpenLineage spec might not contain all the facets you need to write your extractor,
in which case you will have to make your own `custom facets <https://openlineage.io/docs/spec/facets/custom-facets>`_.
More on creating custom facets can be found `here <https://openlineage.io/blog/extending-with-facets/>`_.
+Custom Run Facets
+-----------------
+
+You can inject your own custom facets in the lineage event's run facet using the ``custom_run_facets`` Airflow configuration.
+
+Steps to be taken,
+
+1. Write a function that returns the custom facet. You can write as many custom facet functions as needed.
+2. Register the functions using the ``custom_run_facets`` Airflow configuration.
+
+Once done, Airflow OpenLineage listener will automatically execute these functions during the lineage event generation
+and append their return values to the run facet in the lineage event.
+
+Writing a custom facet function
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+- **Input arguments:** The function should accept the ``TaskInstance`` as an input argument.
+- **Function body:** Perform the logic needed to generate the custom facet. The custom facet should inherit from the ``BaseFacet`` for the ``_producer`` and ``_schemaURL`` to be automatically added for the facet.
+- **Return value:** The custom facet to be added to the lineage event. Return type should be ``None`` or ``dict``. You may choose to return ``None`` or ``{}``, if you do not want to add custom facets for certain criteria.
+
+**Example custom facet function**
+
+.. code-block:: python
+
+ import attrs
+ from airflow.models import TaskInstance
+ from openlineage.client.facet import BaseFacet
+
+
+ @attrs.define(slots=False)
+ class MyCustomRunFacet(BaseFacet):
+ """Define a custom facet."""
+
+ name: str
+ jobState: str
+ uniqueName: str
+ displayName: str
+ dagId: str
+ taskId: str
+ cluster: str
+
+
+ def get_my_custom_facet(task_instance: TaskInstance):
+ job_unique_name = f"TEST.{task_instance.dag_id}.{task_instance.task_id}"
Review Comment:
Addressed in 3455b732e64a70fc3c041390d8f841987c47828f.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1666299656
##########
airflow/providers/openlineage/provider.yaml:
##########
@@ -104,6 +104,13 @@ config:
example: full.path.to.ExtractorClass;full.path.to.AnotherExtractorClass
default: ~
version_added: ~
+ custom_run_facets:
+ description: |
+ Register custom run facet functions by passing a string of semicolon separated full import paths.
+ type: string
+ example: full.path.to.custom_facet_function;full.path.to.another_custom_facet_function
+ default: ''
+ version_added: 1.8.0
Review Comment:
Addressed in c1bc84efc2a454dc5eb4c0e33eb206506b06e445.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1592624751
##########
airflow/providers/openlineage/conf.py:
##########
@@ -65,6 +65,19 @@ def custom_extractors() -> set[str]:
return set(extractor.strip() for extractor in option.split(";") if extractor.strip())
+@cache
+def custom_facet_functions() -> set[str]:
+ """[openlineage] custom_facet_functions."""
+ option = conf.get(_CONFIG_SECTION, "custom_facet_functions", fallback="")
+ if not option:
+ option = os.getenv("OPENLINEAGE_CUSTOM_FACET_FUNCTIONS", "")
Review Comment:
I think good question is whether we can do it without global configs? I think this might be good to be configurable on task/dag/global level, rather than fixed for whole Airflow instance. It goes well with our effort to have lineage be incrementally enabled.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "boring-cyborg[bot] (via GitHub)" <gi...@apache.org>.
boring-cyborg[bot] commented on PR #38982:
URL: https://github.com/apache/airflow/pull/38982#issuecomment-2242711124
Awesome work, congrats on your first merged pull request! You are invited to check our [Issue Tracker](https://github.com/apache/airflow/issues) for additional contributions.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on PR #38982:
URL: https://github.com/apache/airflow/pull/38982#issuecomment-2208102010
Getting some documentation error on `breeze build-docs openlineage`. Trying to debug and fix this.
<img width="988" alt="image" src="https://github.com/apache/airflow/assets/68486238/3a27cddb-334e-441f-a63e-da7ba9eb4e35">
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1666331170
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -78,11 +78,38 @@ def get_job_name(task: TaskInstance) -> str:
def get_custom_facets(task_instance: TaskInstance | None = None) -> dict[str, Any]:
+ from airflow.providers.openlineage.extractors.manager import try_import_from_string
+
custom_facets = {}
# check for -1 comes from SmartSensor compatibility with dynamic task mapping
# this comes from Airflow code
if hasattr(task_instance, "map_index") and getattr(task_instance, "map_index") != -1:
custom_facets["airflow_mappedTask"] = AirflowMappedTaskRunFacet.from_task_instance(task_instance)
+
+ # Append custom run facets by executing the custom_run_facets.
Review Comment:
>It's more general comment but currently get_custom_facets() function is called only once, after task instance has started.
Updated documentation to capture this, 426cc856fd9ec4071df4fcd50138c7f6c2b2252c.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1666320760
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -78,11 +78,38 @@ def get_job_name(task: TaskInstance) -> str:
def get_custom_facets(task_instance: TaskInstance | None = None) -> dict[str, Any]:
+ from airflow.providers.openlineage.extractors.manager import try_import_from_string
+
custom_facets = {}
# check for -1 comes from SmartSensor compatibility with dynamic task mapping
# this comes from Airflow code
if hasattr(task_instance, "map_index") and getattr(task_instance, "map_index") != -1:
custom_facets["airflow_mappedTask"] = AirflowMappedTaskRunFacet.from_task_instance(task_instance)
+
+ # Append custom run facets by executing the custom_run_facets.
+ for custom_facet_func in conf.custom_run_facets():
+ func: type[function] | None = try_import_from_string(custom_facet_func)
+ if not func:
+ log.warning(
+ "OpenLineage is unable to import custom facet function `%s`; will ignore it.",
+ custom_facet_func,
+ )
+ continue
+ facet: dict[str, BaseFacet] = func(task_instance)
Review Comment:
>I noticed that in examples here we are using attrs.asdict(), so we are providing the facets as dicts, but here we actually expect type[BaseFacet]. Could you adjust the typing and examples so that it matches?
We expect that the functions return `dict` or `None`. Addressed the typing error in 3790b7410a667b2cfe691b8a775d3ac0e31c6fc8.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on PR #38982:
URL: https://github.com/apache/airflow/pull/38982#issuecomment-2210305119
> FYI. Sometimes --clean-build flag helps.
Thanks @potiuk, `--clean-build` was helpful to overcome those documentation build errors.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "XD-DENG (via GitHub)" <gi...@apache.org>.
XD-DENG commented on PR #38982:
URL: https://github.com/apache/airflow/pull/38982#issuecomment-2064794678
Hi @eladkal , @anandhimurali is my colleague from the same team and I'm really excited about her 1st PR to Airflow!
Do you have an idea who among the committers may be the best to help review this openlineage-related change?
Any suggestion/input would be greatly appreciated. Thanks a lot!
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1602602899
##########
airflow/providers/openlineage/conf.py:
##########
@@ -65,6 +65,19 @@ def custom_extractors() -> set[str]:
return set(extractor.strip() for extractor in option.split(";") if extractor.strip())
+@cache
+def custom_facet_functions() -> set[str]:
+ """[openlineage] custom_facet_functions."""
+ option = conf.get(_CONFIG_SECTION, "custom_facet_functions", fallback="")
+ if not option:
+ option = os.getenv("OPENLINEAGE_CUSTOM_FACET_FUNCTIONS", "")
Review Comment:
>I think good question is whether we can do it without global configs? I think this might be good to be configurable on task/dag/global level, rather than fixed for whole Airflow instance. It goes well with our effort to have lineage be incrementally enabled.
@mobuchowski , User can choose to `return None` in the custom facet function, if they don't want to add the custom facet for certain tasks/dags. Thinking if the complexity is going to increase if we handle the task/dag/global level settings at config level.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1602650978
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -61,11 +61,20 @@ def get_job_name(task: TaskInstance) -> str:
def get_custom_facets(task_instance: TaskInstance | None = None) -> dict[str, Any]:
+ from airflow.providers.openlineage.extractors.manager import try_import_from_string
+
custom_facets = {}
# check for -1 comes from SmartSensor compatibility with dynamic task mapping
# this comes from Airflow code
if hasattr(task_instance, "map_index") and getattr(task_instance, "map_index") != -1:
custom_facets["airflow_mappedTask"] = AirflowMappedTaskRunFacet.from_task_instance(task_instance)
+
+ # Append custom run facets by executing the custom_facet_functions.
+ for custom_facet_func in conf.custom_facet_functions():
+ func: type[function] = try_import_from_string(custom_facet_func)
+ facet = func(task_instance) if func else None
+ if facet and isinstance(facet, dict):
+ custom_facets.update(facet)
Review Comment:
>Working on adding duplicate check on the custom facet keys and ignoring such duplicate keys by adding logging.
Addressed in 03ed172b6ba140eaad012872c5e0939c16a0e7da
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1683042949
##########
tests/providers/openlineage/utils/test_utils.py:
##########
@@ -482,3 +484,151 @@ def test_get_task_groups_details_nested():
def test_get_task_groups_details_no_task_groups():
assert _get_task_groups_details(DAG("test_dag", start_date=datetime.datetime(2024, 6, 1))) == {}
+
+
+@patch("airflow.providers.openlineage.conf.custom_run_facets", return_value=set())
+def test_get_custom_facets_with_no_function_definition(mock_custom_facet_funcs):
+ sample_ti = TaskInstance(
+ task=EmptyOperator(task_id="test-task", dag=DAG("test-dag")),
Review Comment:
@anandhimurali I believe adding something like this to all places where you create DAG in tests would solve the problem with the last CI check failing and would make it all greenπ
```suggestion
task=EmptyOperator(task_id="test-task", dag=DAG("test-dag", start_date=datetime.datetime(2024, 7, 1))),
```
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1683261420
##########
tests/providers/openlineage/utils/test_utils.py:
##########
@@ -482,3 +484,151 @@ def test_get_task_groups_details_nested():
def test_get_task_groups_details_no_task_groups():
assert _get_task_groups_details(DAG("test_dag", start_date=datetime.datetime(2024, 6, 1))) == {}
+
+
+@patch("airflow.providers.openlineage.conf.custom_run_facets", return_value=set())
+def test_get_custom_facets_with_no_function_definition(mock_custom_facet_funcs):
+ sample_ti = TaskInstance(
+ task=EmptyOperator(task_id="test-task", dag=DAG("test-dag")),
Review Comment:
Thanks again. Addressed in 8c404b93bab1624e8ec893e1fd6b2ff86682a773.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on PR #38982:
URL: https://github.com/apache/airflow/pull/38982#issuecomment-2210548250
@anandhimurali If you need any help to make the CI pass let me know, If you run the pre-commits locally then all of the static checks will pass. When it comes to 2.7 compatibility tests, you simply need to add start_date to the test DAGs you are creating and it should also work 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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1666420076
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -78,11 +78,38 @@ def get_job_name(task: TaskInstance) -> str:
def get_custom_facets(task_instance: TaskInstance | None = None) -> dict[str, Any]:
+ from airflow.providers.openlineage.extractors.manager import try_import_from_string
+
custom_facets = {}
# check for -1 comes from SmartSensor compatibility with dynamic task mapping
# this comes from Airflow code
if hasattr(task_instance, "map_index") and getattr(task_instance, "map_index") != -1:
custom_facets["airflow_mappedTask"] = AirflowMappedTaskRunFacet.from_task_instance(task_instance)
+
+ # Append custom run facets by executing the custom_run_facets.
Review Comment:
>I believe we need to decide whether custom run facets should receive the task before or after execution. Users might want to use elements like query_id generated during execution in their custom facets. While I don't have a strong preference either way, it's important to note this decision consciously in the documentation.
Since the [get_custom_facets](https://github.com/apache/airflow/blob/9d92feda7fae0f014743153ce69656eb339fc420/airflow/providers/openlineage/plugins/listener.py#L166) is invoked only during the OpenLineage START event generation, only the state at the start of the TaskInstance is available to the custom facet functions.
However, if the state after task execution needs to be made available to the functions, it can be only be done during the COMPLETE or other adhoc lineage events? IMO, the scope of changes will be larger for this and can be taken as a follow-up PR. Let me know what 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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on PR #38982:
URL: https://github.com/apache/airflow/pull/38982#issuecomment-2110598999
>Hey @anandhimurali @XD-DENG, is there anything i could help you with here? I really like the idea and would like to see it happen, let me know π
Hi @kacpermuda, Thanks for the interest and I'm excited to hear that. Sorry I've been busy for the last few weeks. I'll look into the comments this week.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on PR #38982:
URL: https://github.com/apache/airflow/pull/38982#issuecomment-2110380257
Hey @anandhimurali @XD-DENG, is there anything i could help you with here? I really like the idea and would like to see it happen, let me know π
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on PR #38982:
URL: https://github.com/apache/airflow/pull/38982#issuecomment-2065139485
Ping @mobuchowski for review :)
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1602589624
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -61,11 +61,20 @@ def get_job_name(task: TaskInstance) -> str:
def get_custom_facets(task_instance: TaskInstance | None = None) -> dict[str, Any]:
+ from airflow.providers.openlineage.extractors.manager import try_import_from_string
+
custom_facets = {}
# check for -1 comes from SmartSensor compatibility with dynamic task mapping
# this comes from Airflow code
if hasattr(task_instance, "map_index") and getattr(task_instance, "map_index") != -1:
custom_facets["airflow_mappedTask"] = AirflowMappedTaskRunFacet.from_task_instance(task_instance)
+
+ # Append custom run facets by executing the custom_facet_functions.
+ for custom_facet_func in conf.custom_facet_functions():
+ func: type[function] = try_import_from_string(custom_facet_func)
+ facet = func(task_instance) if func else None
+ if facet and isinstance(facet, dict):
+ custom_facets.update(facet)
Review Comment:
Working on adding duplicate check on the custom facet keys and ignoring such duplicate keys by adding logging.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on PR #38982:
URL: https://github.com/apache/airflow/pull/38982#issuecomment-2114094133
> I have two points I think it's worth addressing:
> 1. Whether we pass all the necessary info. I think given task_instance we can call get_template_context and get all the data, so this might not be an issue.
> 2. Whether this acts "per DAG/Task" or it's a mechanism that works for all the tasks equally - this might be defined inside the function so it still might not be an issue other than complicating the function a bit more.
>
> Overall I like the solution, but might be good to specify the goals regarding 1) and 2) - because for sure, later, someone will come and ask "how to make this custom facet work only on some particular tasks" and "how do I pass some custom info to my function".
Hi @mobuchowski,
1. Yes, I think `task_instance` is sufficient to fetch necessary info about the run context.
2. The `custom_facet_functions` defined works equally for all tasks. Yes, user can choose to `return None` for scenarios where they won't to add the custom facet.
Let me know if this needs to be documented anywhere. Thanks.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "JDarDagran (via GitHub)" <gi...@apache.org>.
JDarDagran commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1686210312
##########
tests/providers/openlineage/utils/custom_facet_fixture.py:
##########
@@ -0,0 +1,87 @@
+# 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.
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import attrs
+from openlineage.client.facet import BaseFacet
+
+if TYPE_CHECKING:
+ from airflow.models import TaskInstance
+
+
+@attrs.define(slots=False)
+class MyCustomRunFacet(BaseFacet):
+ """Define a custom run facet."""
+
+ name: str
+ jobState: str
+ uniqueName: str
+ displayName: str
+ dagId: str
+ taskId: str
+ cluster: str
+
+
+def get_additional_test_facet(task_instance: TaskInstance) -> dict[str, dict] | None:
+ operator_name = task_instance.task.operator_name if task_instance.task else None
+ if operator_name == "BashOperator":
+ return None
+ job_unique_name = f"TEST.{task_instance.dag_id}.{task_instance.task_id}"
+ return {
+ "additional_run_facet": attrs.asdict(
Review Comment:
nit: same as in docs comment, no need for that
##########
docs/apache-airflow-providers-openlineage/guides/developer.rst:
##########
@@ -446,15 +446,100 @@ Conversion from Airflow Table entity to OpenLineage Dataset is made in the follo
.. _custom_facets:openlineage:
-Custom facets
+Custom Facets
=============
To learn more about facets in OpenLineage, please refer to `facet documentation <https://openlineage.io/docs/spec/facets/>`_.
-Also check out `available Facets <https://github.com/OpenLineage/OpenLineage/blob/main/client/python/openlineage/client/facet.py>`_
+Also check out `available facets <https://github.com/OpenLineage/OpenLineage/blob/main/client/python/openlineage/client/facet.py>`_
The OpenLineage spec might not contain all the facets you need to write your extractor,
in which case you will have to make your own `custom facets <https://openlineage.io/docs/spec/facets/custom-facets>`_.
More on creating custom facets can be found `here <https://openlineage.io/blog/extending-with-facets/>`_.
+Custom Run Facets
+=================
+
+You can inject your own custom facets in the lineage event's run facet using the ``custom_run_facets`` Airflow configuration.
+
+Steps to be taken,
+
+1. Write a function that returns the custom facet. You can write as many custom facet functions as needed.
+2. Register the functions using the ``custom_run_facets`` Airflow configuration.
+
+Once done, Airflow OpenLineage listener will automatically execute these functions during the lineage event generation
+and append their return values to the run facet in the lineage event.
+
+Writing a custom facet function
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+- **Input arguments:** The function should accept the ``TaskInstance`` as an input argument.
+- **Function body:** Perform the logic needed to generate the custom facet. The custom facet should inherit from the ``BaseFacet`` for the ``_producer`` and ``_schemaURL`` to be automatically added for the facet.
+- **Return value:** The custom facet to be added to the lineage event. Return type should be ``dict[str, dict]`` or ``None``. You may choose to return ``None``, if you do not want to add custom facets for certain criteria.
+
+**Example custom facet function**
+
+.. code-block:: python
+
+ import attrs
+ from airflow.models import TaskInstance
+ from openlineage.client.facet import BaseFacet
+
+
+ @attrs.define(slots=False)
+ class MyCustomRunFacet(BaseFacet):
+ """Define a custom facet."""
+
+ name: str
+ jobState: str
+ uniqueName: str
+ displayName: str
+ dagId: str
+ taskId: str
+ cluster: str
+
+
+ def get_my_custom_facet(task_instance: TaskInstance) -> dict[str, dict] | None:
+ operator_name = task_instance.task.operator_name
+ if operator_name == "BashOperator":
+ return
+ job_unique_name = f"TEST.{task_instance.dag_id}.{task_instance.task_id}"
+ return {
+ "additional_run_facet": attrs.asdict(
Review Comment:
nit: there's no need to translate to dict here, OL client Serde takes care of it
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on PR #38982:
URL: https://github.com/apache/airflow/pull/38982#issuecomment-2242711641
Thanks @anandhimurali for implementing this feature π
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1579417082
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -61,11 +61,20 @@ def get_job_name(task: TaskInstance) -> str:
def get_custom_facets(task_instance: TaskInstance | None = None) -> dict[str, Any]:
+ from airflow.providers.openlineage.extractors.manager import try_import_from_string
+
custom_facets = {}
# check for -1 comes from SmartSensor compatibility with dynamic task mapping
# this comes from Airflow code
if hasattr(task_instance, "map_index") and getattr(task_instance, "map_index") != -1:
custom_facets["airflow_mappedTask"] = AirflowMappedTaskRunFacet.from_task_instance(task_instance)
+
+ # Append custom run facets by executing the custom_facet_functions.
+ for custom_facet_func in conf.custom_facet_functions():
+ func: type[function] = try_import_from_string(custom_facet_func)
+ facet = func(task_instance) if func else None
+ if facet and isinstance(facet, dict):
+ custom_facets.update(facet)
Review Comment:
I would add a check for duplicates here (check if we are overwriting some facet already and maybe put some information in logs) and possibly also disallow some predefined facet names (keys in dict), so that some base facets are not overwritten by user accidentally. Or maybe we should allow overwriting and give the user full control? WDYT @mobuchowski @JDarDagran
##########
airflow/providers/openlineage/conf.py:
##########
@@ -65,6 +65,19 @@ def custom_extractors() -> set[str]:
return set(extractor.strip() for extractor in option.split(";") if extractor.strip())
+@cache
+def custom_facet_functions() -> set[str]:
+ """[openlineage] custom_facet_functions."""
+ option = conf.get(_CONFIG_SECTION, "custom_facet_functions", fallback="")
+ if not option:
+ option = os.getenv("OPENLINEAGE_CUSTOM_FACET_FUNCTIONS", "")
Review Comment:
I don't think the `OPENLINEAGE_CUSTOM_FACET_FUNCTIONS` variable is necessary, we can stick to Airflow config. We use `OPENLINEAGE_` variables in some other configs as they are legacy solutions that still need to be supported. For new features we can omit them.
##########
airflow/providers/openlineage/provider.yaml:
##########
@@ -101,6 +101,15 @@ config:
example: full.path.to.ExtractorClass;full.path.to.AnotherExtractorClass
default: ~
version_added: ~
+ custom_facet_functions:
+ description: |
+ Register functions that return custom facets to append to the run facet. This config takes a
+ string of semicolon delimited full import paths to the functions. The functions should accept
+ a TaskInstance input argument and should return a dictionary.
Review Comment:
Not sure if we should include that here as it's an implementation detail. If we change the input parameters it may be easy to forget to change that in the yaml, but maybe I'm wrong?
> The functions should accept a TaskInstance input argument and return a dictionary.
##########
airflow/providers/openlineage/provider.yaml:
##########
@@ -101,6 +101,15 @@ config:
example: full.path.to.ExtractorClass;full.path.to.AnotherExtractorClass
default: ~
version_added: ~
+ custom_facet_functions:
+ description: |
+ Register functions that return custom facets to append to the run facet. This config takes a
+ string of semicolon delimited full import paths to the functions. The functions should accept
+ a TaskInstance input argument and should return a dictionary.
Review Comment:
```suggestion
Register custom facet functions, that return facets to append as run facets, by passing a string of semicolon separated full import paths.
```
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -61,11 +61,20 @@ def get_job_name(task: TaskInstance) -> str:
def get_custom_facets(task_instance: TaskInstance | None = None) -> dict[str, Any]:
+ from airflow.providers.openlineage.extractors.manager import try_import_from_string
+
custom_facets = {}
# check for -1 comes from SmartSensor compatibility with dynamic task mapping
# this comes from Airflow code
if hasattr(task_instance, "map_index") and getattr(task_instance, "map_index") != -1:
custom_facets["airflow_mappedTask"] = AirflowMappedTaskRunFacet.from_task_instance(task_instance)
+
+ # Append custom run facets by executing the custom_facet_functions.
+ for custom_facet_func in conf.custom_facet_functions():
+ func: type[function] = try_import_from_string(custom_facet_func)
Review Comment:
`func` can also be None, so we should maybe include that in type hint?
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on PR #38982:
URL: https://github.com/apache/airflow/pull/38982#issuecomment-2208101080
>I think it would be good to rename this functionality to custom_run_facets, we might need to inject other facets too, specifically Job.facets.
Addressed in 3a11b7625c896252c681bf1f217f0cce1f5eca57.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1666300729
##########
docs/apache-airflow-providers-openlineage/guides/developer.rst:
##########
@@ -446,15 +446,96 @@ Conversion from Airflow Table entity to OpenLineage Dataset is made in the follo
.. _custom_facets:openlineage:
-Custom facets
+Custom Facets
=============
To learn more about facets in OpenLineage, please refer to `facet documentation <https://openlineage.io/docs/spec/facets/>`_.
-Also check out `available Facets <https://github.com/OpenLineage/OpenLineage/blob/main/client/python/openlineage/client/facet.py>`_
+Also check out `available facets <https://github.com/OpenLineage/OpenLineage/blob/main/client/python/openlineage/client/facet.py>`_
The OpenLineage spec might not contain all the facets you need to write your extractor,
in which case you will have to make your own `custom facets <https://openlineage.io/docs/spec/facets/custom-facets>`_.
More on creating custom facets can be found `here <https://openlineage.io/blog/extending-with-facets/>`_.
+Custom Run Facets
+-----------------
+
+You can inject your own custom facets in the lineage event's run facet using the ``custom_run_facets`` Airflow configuration.
+
+Steps to be taken,
+
+1. Write a function that returns the custom facet. You can write as many custom facet functions as needed.
+2. Register the functions using the ``custom_run_facets`` Airflow configuration.
+
+Once done, Airflow OpenLineage listener will automatically execute these functions during the lineage event generation
+and append their return values to the run facet in the lineage event.
+
+Writing a custom facet function
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+- **Input arguments:** The function should accept the ``TaskInstance`` as an input argument.
+- **Function body:** Perform the logic needed to generate the custom facet. The custom facet should inherit from the ``BaseFacet`` for the ``_producer`` and ``_schemaURL`` to be automatically added for the facet.
+- **Return value:** The custom facet to be added to the lineage event. Return type should be ``None`` or ``dict``. You may choose to return ``None`` or ``{}``, if you do not want to add custom facets for certain criteria.
+
+**Example custom facet function**
+
+.. code-block:: python
+
+ import attrs
+ from airflow.models import TaskInstance
+ from openlineage.client.facet import BaseFacet
+
+
+ @attrs.define(slots=False)
+ class MyCustomRunFacet(BaseFacet):
+ """Define a custom facet."""
+
+ name: str
+ jobState: str
+ uniqueName: str
+ displayName: str
+ dagId: str
+ taskId: str
+ cluster: str
+
+
+ def get_my_custom_facet(task_instance: TaskInstance):
+ job_unique_name = f"TEST.{task_instance.dag_id}.{task_instance.task_id}"
Review Comment:
>Including some simple example here can make it easier for the users to add conditional facets.
Addressed in 3455b732e64a70fc3c041390d8f841987c47828f.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "anandhimurali (via GitHub)" <gi...@apache.org>.
anandhimurali commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1666303994
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -78,11 +78,38 @@ def get_job_name(task: TaskInstance) -> str:
def get_custom_facets(task_instance: TaskInstance | None = None) -> dict[str, Any]:
+ from airflow.providers.openlineage.extractors.manager import try_import_from_string
+
custom_facets = {}
# check for -1 comes from SmartSensor compatibility with dynamic task mapping
# this comes from Airflow code
if hasattr(task_instance, "map_index") and getattr(task_instance, "map_index") != -1:
custom_facets["airflow_mappedTask"] = AirflowMappedTaskRunFacet.from_task_instance(task_instance)
+
+ # Append custom run facets by executing the custom_run_facets.
+ for custom_facet_func in conf.custom_run_facets():
+ func: type[function] | None = try_import_from_string(custom_facet_func)
+ if not func:
+ log.warning(
+ "OpenLineage is unable to import custom facet function `%s`; will ignore it.",
+ custom_facet_func,
+ )
+ continue
+ facet: dict[str, BaseFacet] = func(task_instance)
Review Comment:
>And one more question / decision to make: should we wrap the function call with try/except to avoid crashing the whole event emission or do we let the user crash it if they provided the function that breaks? I would vote for putting a try/except with some descriptive logging here.
Makes sense. Added try-except in db25be02a92c2f0f81b27710d92fb760365f2148.
--
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] Ability to add custom facet in OpenLineage events [airflow]
Posted by "kacpermuda (via GitHub)" <gi...@apache.org>.
kacpermuda commented on code in PR #38982:
URL: https://github.com/apache/airflow/pull/38982#discussion_r1666431372
##########
airflow/providers/openlineage/utils/utils.py:
##########
@@ -78,11 +78,38 @@ def get_job_name(task: TaskInstance) -> str:
def get_custom_facets(task_instance: TaskInstance | None = None) -> dict[str, Any]:
+ from airflow.providers.openlineage.extractors.manager import try_import_from_string
+
custom_facets = {}
# check for -1 comes from SmartSensor compatibility with dynamic task mapping
# this comes from Airflow code
if hasattr(task_instance, "map_index") and getattr(task_instance, "map_index") != -1:
custom_facets["airflow_mappedTask"] = AirflowMappedTaskRunFacet.from_task_instance(task_instance)
+
+ # Append custom run facets by executing the custom_run_facets.
+ for custom_facet_func in conf.custom_run_facets():
+ func: type[function] | None = try_import_from_string(custom_facet_func)
+ if not func:
+ log.warning(
+ "OpenLineage is unable to import custom facet function `%s`; will ignore it.",
+ custom_facet_func,
+ )
+ continue
+ facet: dict[str, BaseFacet] = func(task_instance)
Review Comment:
I was more concerned about the type of the dict's key, so whether the function should produce:
`facet: dict[str, Type[BaseFacet]] | None` or `facet: dict[str, dict] | None` ? Maybe it doesn't matter, but then we should probably also let the users know π
--
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