feat: Adjust DBT OpenLineage to Airflow 3 and improve logging #47500
+92
−31
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
kacpermuda
commented
Mar 7, 2025
Comment on lines
+38
to
+40
| if AIRFLOW_V_3_0_PLUS: | ||
| dagrun = task_instance.get_template_context()["dag_run"] | ||
| return dagrun.logical_date or dagrun.run_after |
There was a problem hiding this comment.
@mobuchowski Does this look right? I was trying to mimic OL's listener behavior.
There was a problem hiding this comment.
Yes, given that you don't have to deal with tests without dag_run :)
mobuchowski
reviewed
Mar 7, 2025
| @@ -286,6 +286,10 @@ def get_openlineage_facets_on_complete(self, task_instance) -> OperatorLineage: | |||
|
|
|||
| if isinstance(self.run_id, int) and self.wait_for_termination is True: | |||
| return generate_openlineage_events_from_dbt_cloud_run(operator=self, task_instance=task_instance) | |||
| self.log.info( | |||
| "Extraction of OpenLineage events from DBT will be skipped: " | |||
| "`run_id` is not set OR `self.wait_for_termination` is False. " | |||
There was a problem hiding this comment.
If you're logging this, you might as well display which one of those is true
Comment on lines
+120
to
+125
| except Exception: # type: ignore | ||
| log.info( | ||
| "Openlineage could not find DBT catalog artifact, usually available when docs are generated." | ||
| "Proceeding with metadata extraction. " | ||
| "If you see error logs above about `HTTP error: Not Found` it's safe to ignore them." | ||
| ) |
There was a problem hiding this comment.
If that's true and the feature is optional, we should not log those errors and maybe display one collective error message at the end if possible.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Main change in this PR is adjusting the way
_get_logical_datefunction works, now it should mimic OpenLineage listener's behavior. This is crucial, as OL run_id generated here must match the one generated by the listener, so that the parent-child relationship is correctly created.Additionally, some logs have been added that will be useful for debugging any issues with dbt OL extraction.
cc @mobuchowski
^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named
{pr_number}.significant.rstor{issue_number}.significant.rst, in newsfragments.