Lee-W commented on code in PR #39110:
URL: https://github.com/apache/airflow/pull/39110#discussion_r1573287116
##########
airflow/providers/databricks/operators/databricks.py:
##########
@@ -156,17 +144,37 @@ def
_handle_deferrable_databricks_operator_execution(operator, hook, log, contex
log.info("%s completed successfully.", operator.task_id)
-def _handle_deferrable_databricks_operator_completion(event: dict, log:
Logger) -> None:
+def _get_databricks_notebook_error(run_info: dict, hook: DatabricksHook,
run_state: RunState) -> str:
+ task_run_id = None
+ if "tasks" in run_info:
+ for task in run_info["tasks"]:
+ if task.get("state", {}).get("result_state", "") == "FAILED":
+ task_run_id = task["run_id"]
+ if task_run_id is not None:
+ run_output = hook.get_run_output(task_run_id)
+ if "error" in run_output:
+ notebook_error = run_output["error"]
+ else:
+ notebook_error = run_state.state_message
+ else:
+ notebook_error = run_state.state_message
+ return notebook_error
+
+
+def _handle_deferrable_databricks_operator_completion(event: dict, log:
Logger, hook: DatabricksHook) -> None:
validate_trigger_event(event)
run_state = RunState.from_json(event["run_state"])
run_page_url = event["run_page_url"]
+ run_id = event["run_id"]
log.info("View run status, Spark UI, and logs at %s", run_page_url)
if run_state.is_successful:
log.info("Job run completed successfully.")
return
+ run_info = hook.get_run(run_id)
Review Comment:
I feel like we should do something like
https://github.com/apache/airflow/blob/4374a4987218a05913253b368f45b5fe0ef21493/airflow/providers/databricks/hooks/databricks.py#L432
for deferrable mode. I think it might be possible to implement an async
`hook.get_run_output(task_run_id)`
--
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]
