Catch Metadata Workflow databricks

[jeremy98]

Hello community,

Is it possible to get metadata workflow of a databricks job that is running? Like the start time, end time, triggered by etc.? Using dbutils.widgets.get()?

[Ayushi_Suthar]

Hi @jeremy98 , Good Day! 

Could you please refer to the below community post on how to Retrieve job-level parameters in Python? 

https://community.databricks.com/t5/data-engineering/retrieve-job-level-parameters-in-python/td-p/44...

Also please refer to this document to understand how we can use the command dbutils.widgets.get

https://docs.databricks.com/en/dev-tools/databricks-utils.html#get-command-dbutilswidgetsget

Please let me know if this helps and leave a like if this information is useful, followups are appreciated.
Kudos
Ayushi

[jeremy98]

I want this information about databricks workflow run (metadata):

[fmadeiro]

@jeremy98, 

Accessing metadata such as the start time, end time, and trigger information of a running Databricks job cannot be accomplished using dbutils.widgets.get(). The dbutils.widgets utility is designed for retrieving the current value of input widgets within notebooks and does not provide access to job metadata.

To obtain metadata about a running job, you should utilize the Databricks REST API, specifically the Jobs API. The GET /api/2.0/jobs/runs/get endpoint allows you to retrieve detailed information about a specific job run, including its start time, end time, and the user who triggered the run. See the Jobs Api docs for more deitails Jobs API 2.0 | Databricks on AWS

Here's an example of how you can use the Databricks REST API to retrieve job run metadata:

import requests

# Replace with your Databricks workspace URL and access token
workspace_url = 'https://<databricks-instance>'
access_token = 'your_access_token'

# Replace with your job run ID
run_id = '<run_id>'

# Set up the request headers with the access token
headers = {
    'Authorization': f'Bearer {access_token}'
}

# Make the API request to get job run details
response = requests.get(f'{workspace_url}/api/2.0/jobs/runs/get?run_id={run_id}', headers=headers)

# Check if the request was successful
if response.status_code == 200:
    run_details = response.json()
    # Extract metadata from the response
    start_time = run_details.get('start_time')
    end_time = run_details.get('end_time')
    triggered_by = run_details.get('creator_user_name')
    print(f'Start Time: {start_time}')
    print(f'End Time: {end_time}')
    print(f'Triggered By: {triggered_by}')
else:
    print(f'Error: {response.status_code} - {response.text}')

[jeremy98]

Hello,
Thanks for this answer :), but if I want to know which is the last run about the same workflow that I actually run? How to catch the run_id?

EDIT: ok, I need to change the API url in this way 

/api/2.0/jobs/runs/list?job_id=<id-of-the-job> ... thanks! Considering the query parameters possible to use perfect

[fmadeiro · Accepted solution]

To identify the run_id of the last run of a specific Databricks job (or workflow), you can use the Databricks REST API's GET /api/2.0/jobs/runs/list endpoint. This endpoint returns a list of runs for a specified job, sorted in descending order by their start time. By examining the most recent entry, you can obtain the run_id of the last run.

Here’s how you can catch the run_id of the most recent run for a given job:

import requests

# Replace with your Databricks workspace URL and access token
workspace_url = 'https://<databricks-instance>'
access_token = 'your_access_token'

# Replace with your job ID
job_id = '<job_id>'

# Set up the request headers with the access token
headers = {
    'Authorization': f'Bearer {access_token}'
}

# Make the API request to list runs for the specified job
response = requests.get(f'{workspace_url}/api/2.0/jobs/runs/list?job_id={job_id}', headers=headers)

# Check if the request was successful
if response.status_code == 200:
    runs = response.json().get('runs', [])
    if runs:
        # Get the most recent run (first in the list)
        last_run = runs[0]
        last_run_id = last_run.get('run_id')
        print(f'Last Run ID: {last_run_id}')
    else:
        print('No runs found for the specified job.')
else:
    print(f'Error: {response.status_code} - {response.text}')