topic Re: Need to automatically rerun the failed jobs in databricks in Data Engineering

Need to automatically rerun the failed jobs in databricks

Vishalakshi — Sun, 08 Sep 2024 14:53:19 GMT

Hi all,

I need to retrigger the failed jobs automatically in data bricks, can you please help me with all the possible ways to make it possible

Re: Need to automatically rerun the failed jobs in databricks

ranged_coop — Sun, 08 Sep 2024 15:23:57 GMT

have a look at this link - https://docs.databricks.com/en/jobs/settings.html#retry-policies

you can set retry logic for tasks or have the job run in a loop and check manually for the status and re-run if not successful.

Re: Need to automatically rerun the failed jobs in databricks

filipniziol — Sun, 08 Sep 2024 17:10:34 GMT

Hi @ranged_coop ,

To automatically retrigger failed jobs in Databricks within the last 24 hours, you can use the Databricks REST API to list the jobs, filter out the failed runs, and then retrigger those failed jobs. Below is a Python script that will help you achieve this.

import requests import datetime # Databricks configurations DATABRICKS_HOST = "<your workspace url>" # Replace with your Databricks workspace URL DATABRICKS_TOKEN = "<your personal access token>" # Replace with your Databricks Personal Access Token # API endpoints LIST_RUNS_ENDPOINT = f"{DATABRICKS_HOST}/api/2.1/jobs/runs/list" RUN_NOW_ENDPOINT = f"{DATABRICKS_HOST}/api/2.1/jobs/run-now" # Headers for API requests HEADERS = { "Authorization": f"Bearer {DATABRICKS_TOKEN}" } print(datetime.datetime.now()) # Get the timestamp for 24 hours ago twenty_four_hours_ago = datetime.datetime.now() - datetime.timedelta(hours=24) def get_failed_runs(): """ Retrieve all failed job runs within the last 24 hours. """ failed_runs = [] has_more = True offset = 0 while has_more: # Fetch job runs with pagination response = requests.get(LIST_RUNS_ENDPOINT, headers=HEADERS, params={"offset": offset, "limit": 25}) data = response.json() # Check if data fetch was successful if response.status_code != 200 or "runs" not in data: print(f"Failed to fetch job runs: {data.get('message', 'Unknown error')}") break for run in data["runs"]: # Check if the run failed and was within the last 24 hours run_end_time = datetime.datetime.fromtimestamp(run["end_time"] / 1000) if "state" in run and "result_state" in run["state"] and run["state"]["result_state"] == "FAILED" and run_end_time > twenty_four_hours_ago: failed_runs.append(run) # Check for more runs has_more = data.get("has_more", False) offset += 25 # Increment offset to fetch next set of runs return failed_runs def retrigger_failed_runs(failed_runs): """ Retrigger all failed job runs. """ for run in failed_runs: job_id = run["job_id"] print(f"Retriggering job ID: {job_id}, Run ID: {run['run_id']}") response = requests.post(RUN_NOW_ENDPOINT, headers=HEADERS, json={"job_id": job_id}) if response.status_code == 200: print(f"Successfully retriggered job {job_id}.") else: print(f"Failed to retrigger job {job_id}: {response.json().get('message', 'Unknown error')}") # Main script execution if __name__ == "__main__": failed_runs = get_failed_runs() if failed_runs: print(f"Found {len(failed_runs)} failed runs in the last 24 hours.") retrigger_failed_runs(failed_runs) else: print("No failed runs found in the last 24 hours.")

Re: Need to automatically rerun the failed jobs in databricks

Vishalakshi — Fri, 13 Sep 2024 09:46:26 GMT

I tried with retry logic, but I need to trigger those job if they fail only due to timeout/server issues, Can you help me on this

Re: Need to automatically rerun the failed jobs in databricks

Vishalakshi — Fri, 13 Sep 2024 09:49:14 GMT

Thanks for the script.

But I need to write a code within the existing notebook, that code should filter and retrigger the notebook if it will fail due to any timeout/server issues.

Can you please help me with the script

Re: Need to automatically rerun the failed jobs in databricks

filipniziol — Tue, 17 Sep 2024 16:19:03 GMT

Hi @Vishalakshi ,

I have responded during the weekend, but it seems the responses were lost.
You have here the run object. For example the current criteria is to return only runs where run[state][result_state] == "FAILED" so basically all failed jobs.

What you can do is to print(run) and you will see the object structure:

For example to rerun the jobs where run["state"]["state_message"] contains "Task df_regular failed with message: Workload failed" the code would be:

for run in data["runs"]: # Check if the run failed and was within the last 24 hours run_end_time = datetime.datetime.fromtimestamp(run["end_time"] / 1000) if "state" in run and "result_state" in run["state"] and run["state"]["result_state"] == "FAILED" \ and "Task df_regular failed with message: Workload failed" in run["state"]["state_message"] \ and run_end_time > twenty_four_hours_ago: failed_runs.append(run)

So I recommend just printing the run object, and building the filtering logic according to your criteria.

Hope it helps!