Add base OpenLineage provider implementation #29940

mobuchowski · 2023-03-06T18:34:51Z

This PR consistent mostly of code that was created in OpenLineage project. It consists of

Provider wiring
OpenLineageListener that uses Listener API to get notification about changes to TaskInstance and Dag states
Extractor framework, which is used to extract lineage information from particular operators. It's ment to be replaced by direct implementation of lineage features in later phase and extracting them using DefaultExtractor. This PR does not include actual extractors, but code around using and registering them.
OpenLineageAdapter that translates extracted information to OpenLineage events.
Utils around specific Airflow OL facets and features

This is a base implementation that's not ment to be released yet, but to add code modified to be consistent with Airflow standards, get early feedback and provide canvas to add later features, docs, tests on.

Closes: #29669

airflow/providers/openlineage/extractors/__init__.py

ashb

(Will review in more detail when I can)

airflow/providers/openlineage/extractors/base.py

JDarDagran · 2023-03-09T19:41:05Z

airflow/providers/openlineage/provider.yaml

+  - 1.0.0
+
+dependencies:
+  - apache-airflow>=2.5.1


This needs to be set as the most recent published Airflow version. Otherwise, it breaks on Airflow 2.3 installation and provider verification step. Ideally, we'd want to have it set to the first version released with any core changes needed to make OpenLineage provider working (not sure if it'll be 2.5.2 or 2.6.0)

Not something we should focus on now. I will not release the provider before the AIP is complete so we can update this one later.
OL support is considered a feature thus core changes related to it will be in 2.6 or 2.7

JDarDagran · 2023-03-10T19:30:50Z

Also, Build docs step fails when trying to fetch inventory for OpenLineage provider. I'm not sure if that's something that should be uploaded manually at first?

ashb

I'm not quite sure the split between ExtractorManager and Extractors -- they both seem like they are used together. Maybe they should both live in ol.extractors.manager file.

Why plugins in the airflow.providers.openlineage.plugins package name? How pluggable are they?

I'm not all that sold on the amount of code in the utils __init__.py file (both code in init file along side other files, and the amount of code in general. At the very least lets not have code in there when there is also a converters.py along side it)

Haven't reviewed the tests yet. I'll get to them in a day or two

airflow/providers/openlineage/extractors/base.py

ashb · 2023-03-13T13:37:03Z

airflow/providers/openlineage/extractors/base.py

+        Return the connection URI for the given ID. We first attempt to lookup
+        the connection URI via AIRFLOW_CONN_<conn_id>, else fallback on querying
+        the Airflow's connection table.


Firstly: This doc string doesn't really match the implementation.

Secondly: How does this differ from just conn.get_uri() It's not clear from reading the function alone. Please add some comments saying why that isn't suitable (or is it?)

Changed this to redacted_connection_uri and added appropriate comment.

This method uses conn.get_uri() to get connection URI, then removes potential username, password, and known sensitive query parts. I've moved passing that second part to being provided by extractor/operator that calls it.

ashb · 2023-03-13T13:39:00Z

airflow/providers/openlineage/extractors/extractors.py

+        # Extractors should implement BaseExtractor
+        from airflow.providers.openlineage.utils import import_from_string
+
+        env_extractors = os.getenv("OPENLINEAGE_EXTRACTORS")


Possibly this should now be conf.get("openlineage", "extractors") or similar now.

Actually, we have a series of issues opened linked to AIP-53, one of them is #29670. Along with that we'd like to integration OpenLineage configuration into Airflow config.
Besides that, it's also a matter of backwards compatibility for openlineage-airflow (current OL-Airflow integration outside of Airflow provider) in which you could put custom extractors in this env var.

ashb · 2023-03-13T13:39:26Z

airflow/providers/openlineage/extractors/extractors.py

+
+        env_extractors = os.getenv("OPENLINEAGE_EXTRACTORS")
+        if env_extractors is not None:
+            for extractor in env_extractors.split(";"):


Comma more often used for this purpose in Airflow

; was supported in openlineage-airflow. I think that for a while we'd want to maintain support of custom extractors as it was. Eventually, we could split on ; or ,.

airflow/providers/openlineage/utils/__init__.py

airflow/providers/openlineage/utils/converters.py

bolkedebruin · 2023-03-17T10:21:39Z

Can I say that this integration needs extensive docs and architecture description (inside those docs)? It is quite opaque to me know how this works, why it needs workers at all, what it does to my running system to have those workers, how does it affect task runs, what if the code fails did my task fail? etc etc.

potiuk · 2023-03-17T11:10:20Z

Can I say that this integration needs extensive docs and architecture description (inside those docs)? It is quite opaque to me know how this works, why it needs workers at all, what it does to my running system to have those workers, how does it affect task runs, what if the code fails did my task fail? etc etc.

Yeah. I think there are quite a number of things here to make decisions on. I am also going to have a closer look and make a more deep review after the presentation next week, I hope Iy will have much more context after seeing some of the decisions and reasoning for the open-lineage architecture there. I recall how useful it was to get a walkthrough by @ashb of the new scheduler architecture and decisions back in the 2.0 days (and then seeing the "Deep dive" talk from the summit) - this allowed those who participated/watched (and paid attention :) ) to better reason in case of future issues/questions and be able to fix problem or diagnose issues or to propose improvements (even though some of the details there are a bit arcane). I wish for example we had something like that for the Celery Executor or K8S integration when it comes to stalling, adoption, log streaming etc..

That's also note for @mobuchowski and @julienledem. The more of those decisions and contex will be explained, documented (including recording and publishing the meeting is a good idea - ideally followed up by a talk on the Summit) the more it will be a community effort.

For example I am planning to submit a talk for the summit with a working title "Everything you even did not know you wanted to ask about the Airflow CI (or was terrified to ask)" to address at least part of the SPOF problem we have there and pave the way to get others at least being able to reason on where to fix when there are problems.

potiuk · 2023-03-17T11:15:27Z

BTW. @bolkedebruin I think the docs is not nearly enough (and sometimes too much docs is overwhelming). The CI of ours is - I think - documented to the maximum extent possible - including state diagrams, context why we are doing what we are doing, architecture decision records describing the decisions, reference of parameters etc. etc. Still I am the SPOF there big time.

ashb · 2023-04-06T13:26:03Z

Lets see if we can sort out the config masking/redacting - possibly by subclassing in someway, something like this:

class OLReadactor(SecretsMasker):
    @classmethod
    def from_masker(cls, other: SecretsMasker):
        instance = cls()
        instance.patterns = other.patterns
        instance.replacer = other.replacer

    def _redact(self, item: Redactable, name: str | None, depth: int, max_depth: int) -> Redacted:
        # custom type checks here
        return super()._redact(item, name, depth, max_depth)

    def _redact_all(self, item: Redactable, depth: int, max_depth: int) -> Redacted:
        # custom type checks here
        return super()._redact_all(item, depth, max_depth)

(Not sure if you'd need both redact and redact_all or if just redact would be enough)

ashb

A good first start -- not ready to release as a provider yet, but lets get this merged and then fix up all the identified issues (which are already created in the project)

docs/spelling_wordlist.txt

eladkal · 2023-04-07T16:21:55Z

airflow/providers/openlineage/extractors/base.py

+        Particularly, in Airflow 2 some operators are deprecated and simply subclass the new
+        implementation, for example airflow.contrib.operators.bigquery_operator.BigQueryOperator.
+        The BigQueryExtractor needs to work with both of them.


Please explain this.
There is no BigQueryOperator in the code base.

Deprecated operators and new versions of operators are backward compatible. They should not have special treatment.

There is no BigQueryOperator in the code base.

I see, it was removed in September. Removed reference to it, just explaining that single extractor can work for multiple operators.

Signed-off-by: Maciej Obuchowski <obuchowski.maciej@gmail.com> Signed-off-by: Jakub Dardzinski <kuba0221@gmail.com> Co-authored-by: Jakub Dardzinski <kuba0221@gmail.com> Fix provider yaml. Signed-off-by: Jakub Dardzinski <kuba0221@gmail.com> Add OpenLineage to extra packages reference. Signed-off-by: Jakub Dardzinski <kuba0221@gmail.com> Change Airflow dependency to 2.5.1+. Signed-off-by: Jakub Dardzinski <kuba0221@gmail.com>

This PR consistent mostly of code that was created in OpenLineage project. It consists of - Provider wiring - OpenLineageListener that uses Listener API to get notification about changes to TaskInstance and Dag states - Extractor framework, which is used to extract lineage information from particular operators. It's ment to be replaced by direct implementation of lineage features in later phase and extracting them using DefaultExtractor. This PR does not include actual extractors, but code around using and registering them. - OpenLineageAdapter that translates extracted information to OpenLineage events. - Utils around specific Airflow OL facets and features This is a base implementation that's not ment to be released yet, but to add code modified to be consistent with Airflow standards, get early feedback and provide canvas to add later features, docs, tests on. Signed-off-by: Jakub Dardzinski <kuba0221@gmail.com>

boring-cyborg bot added area:dev-tools area:plugins area:providers kind:documentation labels Mar 6, 2023

mobuchowski requested review from eladkal, jedcunningham and potiuk March 6, 2023 18:35

mobuchowski added the provider:openlineage AIP-53 label Mar 6, 2023

JDarDagran force-pushed the aip-53-pr branch from 1485664 to 43b115f Compare March 7, 2023 08:33

ashb reviewed Mar 7, 2023

View reviewed changes

airflow/providers/openlineage/extractors/__init__.py Outdated Show resolved Hide resolved

ashb reviewed Mar 7, 2023

View reviewed changes

airflow/providers/openlineage/extractors/base.py Outdated Show resolved Hide resolved

airflow/providers/openlineage/extractors/base.py Outdated Show resolved Hide resolved

mobuchowski force-pushed the aip-53-pr branch 3 times, most recently from 879ec51 to 412fbea Compare March 8, 2023 23:09

JDarDagran force-pushed the aip-53-pr branch 2 times, most recently from c4dd0c4 to f5a709d Compare March 9, 2023 19:38

JDarDagran reviewed Mar 9, 2023

View reviewed changes

ashb requested changes Mar 13, 2023

View reviewed changes

mobuchowski force-pushed the aip-53-pr branch 7 times, most recently from 8d02baa to 04a02b0 Compare March 14, 2023 11:48

mobuchowski force-pushed the aip-53-pr branch from 04a02b0 to 58fd073 Compare March 17, 2023 11:23

mobuchowski force-pushed the aip-53-pr branch 2 times, most recently from eca360d to 8cf2a67 Compare April 6, 2023 09:19

ashb mentioned this pull request Apr 6, 2023

Allow specifying a max_depth to the redact() call #30505

Merged

mobuchowski force-pushed the aip-53-pr branch 2 times, most recently from 4e0e2ab to 80aa0b6 Compare April 6, 2023 15:53

ashb approved these changes Apr 7, 2023

View reviewed changes

mobuchowski force-pushed the aip-53-pr branch from 80aa0b6 to 5b8340e Compare April 7, 2023 16:11

eladkal reviewed Apr 7, 2023

View reviewed changes

mobuchowski force-pushed the aip-53-pr branch 9 times, most recently from a79868c to b1b029b Compare April 13, 2023 13:34

mobuchowski force-pushed the aip-53-pr branch from b1b029b to 55fb750 Compare April 14, 2023 08:41

mobuchowski requested a review from kaxil as a code owner April 14, 2023 08:41

ashb merged commit 55963de into apache:main Apr 14, 2023

JDarDagran mentioned this pull request Apr 18, 2023

Fix when OpenLineage plugins has listener disabled. #30708

Merged

ephraimbuddy added this to the Airflow 2.7.0 milestone May 8, 2023

ephraimbuddy added the changelog:skip Changes that should be skipped from the changelog (CI, tests, etc..) label May 8, 2023

JDarDagran deleted the aip-53-pr branch July 28, 2023 16:47

eladkal mentioned this pull request Jul 29, 2023

Status of testing Providers that were prepared on July 29, 2023 #32936

Closed

69 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add base OpenLineage provider implementation #29940

Add base OpenLineage provider implementation #29940

mobuchowski commented Mar 6, 2023 •

edited

Loading

ashb left a comment

JDarDagran Mar 9, 2023 •

edited

Loading

eladkal Mar 10, 2023

JDarDagran commented Mar 10, 2023

ashb left a comment

ashb Mar 13, 2023

mobuchowski Mar 14, 2023

ashb Mar 13, 2023

JDarDagran Mar 13, 2023

ashb Mar 13, 2023

JDarDagran Mar 13, 2023 •

edited

Loading

bolkedebruin commented Mar 17, 2023

potiuk commented Mar 17, 2023 •

edited

Loading

potiuk commented Mar 17, 2023 •

edited

Loading

ashb commented Apr 6, 2023

ashb left a comment

eladkal Apr 7, 2023 •

edited

Loading

mobuchowski Apr 11, 2023

Add base OpenLineage provider implementation #29940

Add base OpenLineage provider implementation #29940

Conversation

mobuchowski commented Mar 6, 2023 • edited Loading

ashb left a comment

Choose a reason for hiding this comment

JDarDagran Mar 9, 2023 • edited Loading

Choose a reason for hiding this comment

eladkal Mar 10, 2023

Choose a reason for hiding this comment

JDarDagran commented Mar 10, 2023

ashb left a comment

Choose a reason for hiding this comment

ashb Mar 13, 2023

Choose a reason for hiding this comment

mobuchowski Mar 14, 2023

Choose a reason for hiding this comment

ashb Mar 13, 2023

Choose a reason for hiding this comment

JDarDagran Mar 13, 2023

Choose a reason for hiding this comment

ashb Mar 13, 2023

Choose a reason for hiding this comment

JDarDagran Mar 13, 2023 • edited Loading

Choose a reason for hiding this comment

bolkedebruin commented Mar 17, 2023

potiuk commented Mar 17, 2023 • edited Loading

potiuk commented Mar 17, 2023 • edited Loading

ashb commented Apr 6, 2023

ashb left a comment

Choose a reason for hiding this comment

eladkal Apr 7, 2023 • edited Loading

Choose a reason for hiding this comment

mobuchowski Apr 11, 2023

Choose a reason for hiding this comment

mobuchowski commented Mar 6, 2023 •

edited

Loading

JDarDagran Mar 9, 2023 •

edited

Loading

JDarDagran Mar 13, 2023 •

edited

Loading

potiuk commented Mar 17, 2023 •

edited

Loading

potiuk commented Mar 17, 2023 •

edited

Loading

eladkal Apr 7, 2023 •

edited

Loading