mobuchowski commented on code in PR #73655:
URL: https://github.com/apache/airflow/pull/73655#discussion_r4121333876
##########
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:
Can we also raise the requirements in pyproject.toml for those providers too?
##########
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:
Can we have some database system emitted here - an equivalent of
`databricks` or `snowflake` here - I believe the query event itself won't have
that information?
##########
providers/openlineage/src/airflow/providers/openlineage/utils/sql_hook_lineage.py:
##########
@@ -194,7 +194,7 @@ def _get_namespace(hook_id: int) -> str | None:
events.extend(
_create_ol_event_pair(
task_instance=task_instance,
-
job_name=f"{task_instance.dag_id}.{task_instance.task_id}.query.{query_count}",
+
job_name=f"{task_instance.dag_id}.{task_instance.task_id}.hll_query.{query_count}",
Review Comment:
Why are we changing the name here?
--
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]
