Working with interactive clusters in Databricks makes it possible to manually install libraries using the workspace UI. It’s done once and for all. You don’t have to worry about it anymore.
Creating job clusters is another story. You must take care of library installations prior to executing the notebooks which reference these libraries. In the case you’re using Azure Data Factory to orchestrate the whole process you’re lucky, because appending libraries to job clusters is an out-of-the-box functionality. For all other scenarios using the Databricks REST API is one possible option. In fact, you can do this right from a Python notebook. That’s what I’m going to demonstrate in the following lines.
Let’s have a look at the REST API documentation first. We’ll be using the Cluster Status and Install endpoints only. For installing a library, we need to provide the library source and its properties. We need to create a proper HTTP request body in JSON format including the library source and properties. Here’s one example:
{
"pypi": {
"package": "simplejson",
"repo": "https://my-pypi-mirror.com"
}
}
Here “pypi” is the source and {“package”: “simplejson”, “repo”: “https://my-pypi-mirror.com”} are its properties. For more examples check the documentation here.
Let’s create a new Python notebook and then some functions in it. The code is simple and self-explanatory. Comments are available where needed.
We’ll be using only json and requests libraries:
import json import requests from requests.auth import HTTPBasicAuth
Now, let’s create a REST API wrapper function:
def adbAPI(endPoint, body, method = "GET", region, token): """Execute HTTP request against Databricks REST API 2.0""" domain = region + ".azuredatabricks.net" baseURL = "https://%s/api/" % (domain) if method.upper() == "GET": response = requests.get( baseURL + endPoint , auth = HTTPBasicAuth("token", token) , json = body ) else: response = requests.post( baseURL + endPoint , auth = HTTPBasicAuth("token", token) , json = body ) return response
As parameters we’ll take the API endpoint, HTTP request body, HTTP method (GET or POST), Databricks workspace region (westeurope, northeurope, etc.) and, finally, a Databricks token.
Next is a helper function for translating the library status response into a human readable format:
def describeClusterLibraryState(source, clusterId, status): """Converts cluster library status response to a verbose message""" result_map = { "NOT_INSTALLED" : "{} library is not installed on cluster {}.".format(source.title(), clusterId) , "INSTALLED" : "{} library is already installed on cluster {}.".format(source.title(), clusterId) , "PENDING" : "Pending installation of {} library on cluster {} . . .".format(source, clusterId) , "RESOLVING" : "Retrieving metadata for the installation of {} library on cluster {} . . .".format(source, clusterId) , "INSTALLING" : "Installing {} library on cluster {} . . .".format(source, clusterId) , "FAILED" : "{} library failed to install on cluster {}.".format(source.title(), clusterId) , "UNINSTALL_ON_RESTART": "{} library installed on cluster {} has been marked for removal upon cluster restart.".format(source.title(), clusterId) } return result_map[status.upper()]
The parameters here include the library source, the cluster ID and the API’s status response.
Now let’s get to the serious business. Obviously, we need a way to check the current state of the library to see if it’s not already installed or if something went wrong during installation:
def getClusterLibraryStatus(source, properties, dbRegion, dbToken, verbose = True): """Gets the current library status """ source = source.lower() # Validate input assert source in ("jar", "egg", "whl", "pypi", "maven", "cran"), "Error: Invalid library source specified. Valid sources are: jar, egg, whl, pypi, maven, cran" assert properties is not None, "Error: Empty properties provided" # Get the cluster ID from the Spark environment clusterId = spark.conf.get("spark.databricks.clusterUsageTags.clusterId") # Set default result to not installed result = describeClusterLibraryState(source, clusterId, "NOT_INSTALLED") if verbose else "NOT_INSTALLED" # Execute REST API request to get the library statuses libStatuses = adbAPI("2.0/libraries/cluster-status?cluster_id=" + clusterId, None, "GET", dbRegion, dbToken) if libStatuses.status_code == 200: statuses = libStatuses.json() if "library_statuses" in statuses: for status in statuses["library_statuses"]: if status["library"] == {source: properties}: if verbose: result = describeClusterLibraryState(source, clusterId, status["status"]) else: result = status["status"] else: print(status) return result
Parameters here include the library source and properties, Databricks region and token and a verbose flag, which if set to true (default) will make the function return human readable library status.
Finally, we have to create the most important function that will actually install the library:
def installClusterLibrary(source, properties, dbRegion, dbToken): """ Installs a cluster library given correct source and properties are provided For examples see https://docs.databricks.com/api/latest/libraries.html#install """ source = source.lower() # Validate input assert source in ("jar", "egg", "whl", "pypi", "maven", "cran"), "Error: Invalid library source specified. Valid sources are: jar, egg, whl, pypi, maven, cran" assert properties is not None, "Error: Empty properties provided" # Get the cluster ID from the Spark environment clusterId = spark.conf.get("spark.databricks.clusterUsageTags.clusterId") status = getClusterLibraryStatus(source, properties, dbRegion, dbToken, False).upper() if status != "INSTALLED": # Create the HTTP request body based on the cluster ID, library source and properties body = json.loads('{"cluster_id": "' + clusterId + '", "libraries": [{"' + source + '": ' + json.dumps(properties) + '}]}') # Execute REST API request to install the library runAPI = adbAPI("2.0/libraries/install", body, "POST", dbRegion, dbToken) if runAPI.status_code == 200: print("Installation started . . .") else: print(runAPI) else: print(describeClusterLibraryState(source, clusterId, status)) return status
The parameters here are: the library source and properties and the Databricks region and token.
Example function calls (azure-sqldb-spark library hosted on Maven)
Show the library status:
print(getClusterLibraryStatus("maven", {"coordinates": "com.microsoft.azure:azure-sqldb-spark:1.0.2"}, "westeurope", "dapi********************************"))
Install the library:
installClusterLibrary("maven", {"coordinates": "com.microsoft.azure:azure-sqldb-spark:1.0.2"}, "westeurope", "dapi********************************")
One thing to note here is that the library installation starts and runs asynchronously. This means that you can execute the getClusterLibraryStatus function multiple times after installClusterLibrary and get different results until the installation is done (or failed).
Meet the Team – Jason Bonello, Senior Consultant
Meet Jason Bonello! Jason has been with us for just over two years and works
Apr
Meet the Team – Matt How, Principal Consultant
Next up in our series of meet the team blogs is Matt How. Matt has
Apr
MLFlow: Introduction to MLFlow Tracking
MLFlow is an open-source MLOps platform designed by Databricks to enable organisations to easily manage
Apr
Adatis are pleased to announce expansion plans into India
Adatis has offices in London, Surrey and Bulgaria – and has some big expansion plans
Mar
Querying and Importing Data from Excel to SSMS
Introduction There are couple of ways to import data from Excel to SSMS – via
Mar
Data Engineering for Graduates and New Starters
This blog aims to give potential graduates and other new starters in the industry some
Mar
Passing DP-900 Azure Data Fundamentals
Back in December, I took the DP-900 Azure Data Fundamentals exam which is one of
Feb
Real-time Dashboards Provide Transparency to Everyone
Real-time dashboards enable data analytics firm Adatis to be agile and transparent with its team
Feb