kacpermuda commented on code in PR #73655:
URL: https://github.com/apache/airflow/pull/73655#discussion_r4256497972
##########
providers/databricks/src/airflow/providers/databricks/utils/openlineage.py:
##########
@@ -300,36 +248,29 @@ def emit_openlineage_events_for_databricks_queries(
query_metadata if query_metadata else "not found",
)
- query_specific_run_facets: dict[str, RunFacet] = {
- "externalQuery": ExternalQueryRunFacet(externalQueryId=query_id,
source=query_source_namespace)
- }
- if query_metadata.get("error_message"):
- query_specific_run_facets["error"] = ErrorMessageRunFacet(
- message=query_metadata["error_message"],
- programmingLanguage="SQL",
- )
-
- query_specific_job_facets = {}
+ query_job_facets = dict(additional_job_facets or {})
if query_metadata.get("query_text"):
- query_specific_job_facets["sql"] =
SQLJobFacet(query=query_metadata["query_text"])
-
- log.debug("Creating OpenLineage event pair for query ID: %s", query_id)
- event_batch = _create_ol_event_pair(
- job_namespace=namespace(),
-
job_name=f"{task_instance.dag_id}.{task_instance.task_id}.query.{counter}",
+ query_job_facets["sql"] =
SQLJobFacet(query=query_metadata["query_text"])
+
+ log.debug("Emitting OpenLineage event pair for query ID: %s", query_id)
+ emit_query_lineage(
Review Comment:
We do have the namespace in the externalQuery facet, that's either
`databricks://<host etc.>` or `snowflake://<host etc.>`. Not sure if we have
any other facet that is better suited for that.
##########
providers/databricks/src/airflow/providers/databricks/utils/openlineage.py:
##########
@@ -153,44 +145,15 @@ def _get_queries_details_from_databricks(
return query_details
-def _create_ol_event_pair(
- job_namespace: str,
- job_name: str,
- start_time: datetime.datetime,
- end_time: datetime.datetime,
- is_successful: bool,
- run_facets: dict | None = None,
- job_facets: dict | None = None,
-) -> tuple[RunEvent, RunEvent]:
- """Create a pair of OpenLineage RunEvents representing the start and end
of a query execution."""
- from openlineage.client.event_v2 import Job, Run, RunEvent, RunState
- from openlineage.client.uuid import generate_new_uuid
-
- run = Run(runId=str(generate_new_uuid()), facets=run_facets or {})
- job = Job(namespace=job_namespace, name=job_name, facets=job_facets or {})
-
- start = RunEvent(
- eventType=RunState.START,
- eventTime=start_time.isoformat(),
- run=run,
- job=job,
- )
- end = RunEvent(
- eventType=RunState.COMPLETE if is_successful else RunState.FAIL,
- eventTime=end_time.isoformat(),
- run=run,
- job=job,
- )
- return start, end
-
-
-@require_openlineage_version(provider_min_version="2.5.0")
+@require_openlineage_version(provider_min_version="2.16.0")
Review Comment:
Sure !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
