Compartir a través de


SDK de Databricks para Python

Nota:

Databricks recomienda Conjuntos de recursos de Databricks para crear, desarrollar, implementar y probar trabajos y otros recursos de Databricks como código fuente. Consulte ¿Qué son las agrupaciones de recursos de Databricks?

En este artículo, aprenderá a automatizar las operaciones de Azure Databricks y a acelerar el desarrollo con el SDK de Databricks para Python. Este artículo complementa la documentación del SDK de Databricks para Python en Leer los documentos y los ejemplos de códigoen el repositorio del SDK de Databricks para Python en GitHub.

Nota:

El SDK de Databricks para Python está en beta y está bien usarlo en producción.

Durante el período beta, Databricks recomienda anclar una dependencia en la versión secundaria específica del SDK de Databricks para Python de la que depende el código. Por ejemplo, puede anclar dependencias en archivos como requirements.txt para venvo pyproject.toml y poetry.lock para Poesía. Para obtener más información sobre la asignación de dependencias, consulte entornos virtuales y paquetes para venv o Instalación de dependencias para Poesía.

Antes de empezar

Puede usar el SDK de Databricks para Python desde un cuaderno de Azure Databricks o desde la máquina de desarrollo local.

Antes de empezar a usar el SDK de Databricks para Python, la máquina de desarrollo debe cumplir con lo siguiente:

  • Debe tener configurada la autenticación de Azure Databricks.
  • Python 3.8 o superior instalado. Para automatizar los recursos de proceso de Azure Databricks, Databricks recomienda que tenga instaladas las versiones principales y secundarias de Python que coincidan con la instalada en el recurso de proceso de Azure Databricks de destino. Los ejemplos de este artículo se basan en la automatización de clústeres con Databricks Runtime 13.3 LTS, que tiene Instalado Python 3.10. Para obtener la versión correcta, consulte Versiones y compatibilidad de las notas de la versión de Databricks Runtime para la versión de Databricks Runtime del clúster.
  • Databricks recomienda crear y activar un entorno virtual de Python para cada proyecto de Python que use con el SDK de Databricks para Python. Los entornos virtuales de Python ayudan a asegurarse de que el proyecto de código use versiones compatibles de Python y paquetes de Python (en este caso, el SDK de Databricks para el paquete Python). Para más información sobre los entornos virutales de Python, consulta venv o Poetry.

Introducción al SDK de Databricks para Python

En esta sección se describe cómo empezar a trabajar con el SDK de Databricks para Python desde la máquina de desarrollo local. Para usar el SDK de Databricks para Python desde un cuaderno de Azure Databricks, vaya directamente a Uso del SDK de Databricks para Python desde un cuaderno de Azure Databricks.

  1. En la máquina de desarrollo con la autenticación de Azure Databricks configurada, Python ya instalado y el entorno virtual de Python ya activado, instale el paquete de databricks-sdk (y sus dependencias) desde el índice de paquetes de Python (PyPI), como se indica a continuación:

    Venv

    Use pip para instalar el paquete databricks-sdk. (En algunos sistemas, es posible que tenga que reemplazar pip3 por pip, aquí y en todos los lugares).

    pip3 install databricks-sdk
    

    Poetry

    poetry add databricks-sdk
    

    Para instalar una versión específica del paquete databricks-sdk mientras el SDK de Databricks para Python está en versión beta, consulte el historial de versiones del paquete. Por ejemplo, para instalar la versión 0.1.6:

    Venv

    pip3 install databricks-sdk==0.1.6
    

    Poetry

    poetry add databricks-sdk==0.1.6
    

    Para actualizar una instalación existente del paquete del SDK de Databricks para Python a la versión más reciente, ejecute el siguiente comando:

    Venv

    pip3 install --upgrade databricks-sdk
    

    Poetry

    poetry add databricks-sdk@latest
    

    Para mostrar el SDK de Databricks para el paquete de Python actual Version y otros detalles, ejecute el siguiente comando:

    Venv

    pip3 show databricks-sdk
    

    Poetry

    poetry show databricks-sdk
    
  2. En el entorno virtual de Python, cree un archivo de código de Python que importe el SDK de Databricks para Python. En el ejemplo siguiente, en un archivo denominado main.py con el contenido siguiente, basta con enumerar todos los clústeres del área de trabajo de Azure Databricks:

    from databricks.sdk import WorkspaceClient
    
    w = WorkspaceClient()
    
    for c in w.clusters.list():
      print(c.cluster_name)
    
  3. Ejecute el archivo de código Python, para ejecutar el archivo denominado main.py, ejecutando el comando python:

    Venv

    python3.10 main.py
    

    Poetry

    Si está en el shell del entorno virtual:

    python3.10 main.py
    

    Si no está en el shell del entorno virtual:

    poetry run python3.10 main.py
    

    Nota:

    Al no establecer argumentos en la llamada anterior a w = WorkspaceClient(), el SDK de Databricks para Python utiliza su proceso predeterminado para intentar realizar la autenticación de Azure Databricks. Para invalidar este comportamiento predeterminado, consulte la siguiente sección de autenticación.

Autenticación del SDK de Databricks para Python con su cuenta o área de trabajo de Azure Databricks

En esta sección se describe cómo autenticar el SDK de Databricks para Python desde la máquina de desarrollo local a la cuenta o área de trabajo de Azure Databricks. Para usar el SDK de Databricks para Python desde un cuaderno de Azure Databricks, vaya directamente a Uso del SDK de Databricks para Python desde un cuaderno de Azure Databricks.

El SDK de Databricks para Python implementa el estándar de la Autenticación unificada del cliente de Databricks, un enfoque arquitectónico y programático consolidado y coherente para la autenticación. Este enfoque ayuda a configurar y automatizar la autenticación con Azure Databricks de manera más centralizada y predecible. Permite configurar la autenticación de Databricks una vez y, a continuación, usar esa configuración en varias herramientas y SDK de Databricks sin cambios adicionales en la configuración de autenticación. Para obtener más información, incluidos ejemplos de código más completos en Python, consulte Autenticación unificada del cliente de Databricks.

Nota:

El SDK de Databricks para Python aún no ha implementado la autenticación de identidades administradas de Azure.

Entre los patrones de codificación disponibles para inicializar la autenticación de Databricks con el SDK de Databricks para Python se incluyen los siguientes:

  • Para usar la autenticación predeterminada de Databricks, siga uno de estos procedimientos:

    • Cree o identifique un perfil de configuración de Databricks personalizado con los campos necesarios para el tipo de autenticación de Databricks de destino. Luego establezca la variable de entorno DATABRICKS_CONFIG_PROFILE con el nombre del perfil de configuración personalizado.
    • Establezca las variables de entorno necesarias para el tipo de autenticación de Databricks de destino.

    A continuación, cree una instancia de un objeto WorkspaceClient, por ejemplo, con la autenticación predeterminada de Databricks como se indica a continuación:

    from databricks.sdk import WorkspaceClient
    
    w = WorkspaceClient()
    # ...
    
  • Aunque se admite la codificación rígida de los campos obligatorios, no se recomienda porque se corre el riesgo de exponer información confidencial en el código, como los tokens de acceso personal de Azure Databricks. En el ejemplo siguiente se integran como parte del código los valores de host y token de acceso de Azure Databricks para la autenticación de tokens de Databricks:

    from databricks.sdk import WorkspaceClient
    
    w = WorkspaceClient(
      host  = 'https://...',
      token = '...'
    )
    # ...
    

Consulte también Autenticación en la documentación del SDK de Databricks para Python.

Uso del SDK de Databricks para Python desde un cuaderno de Azure Databricks

Puede llamar a la funcionalidad del SDK de Databricks para Python desde un cuaderno de Azure Databricks con un clúster de Azure Databricks asociado con el SDK de Databricks para Python instalado. Está instalado de manera predeterminada en todos los clústeres de Azure Databricks que usan Databricks Runtime 13.3 LTS o una versión superior. Para los clústeres de Azure Databricks que usan Databricks Runtime 12.2 LTS y versiones posteriores, primero debe instalar el SDK de Databricks para Python. Vea Paso 1: Instalación o actualización del SDK de Databricks para Python.

A fin de ver la versión del SDK de Databricks para Python instalado para una versión específica de Databricks Runtime, consulta la sección Bibliotecas de Python instaladas de las notas de la versión de Databricks Runtime para esa versión.

Databricks recomienda instalar la versión más reciente disponible del SDK desde PiPy, pero al menos instalar o actualizar al SDK de Databricks para Python 0.6.0 o superior, ya que la autenticación predeterminada de Notebook de Azure Databricks se usa en la versión 0.6.0 y posteriores en todas las versiones de Databricks Runtime.

Nota:

Databricks Runtime 15.1 es el primer entorno de ejecución de Databricks para tener instalada una versión del SDK de Databricks para Python (0.20.0) que admite la autenticación predeterminada de cuadernos sin que se requiera ninguna actualización.

En la tabla siguiente se describe la compatibilidad de autenticación de cuadernos con el SDK de Databricks para las versiones de Python y Databricks Runtime:

SDK/DBR 10.4 LTS 11.3 LTS 12.3 LTS 13.3 LTS 14.3 LTS 15.1 y versiones posteriores
0.1.7 y versiones anteriores
0.1.10
0.6.0
0.20.0 y versiones posteriores

La autenticación predeterminada de cuadernos de Azure Databricks se basa en un token de acceso personal temporal de Azure Databricks que Azure Databricks genera automáticamente en segundo plano para su propio uso. Azure Databricks elimina este token temporal después de que el cuaderno deje de ejecutarse.

Importante

  • La autenticación predeterminada del cuaderno de Azure Databricks solo funciona en el nodo del controlador del clúster y no en ninguno de los nodos de trabajo o ejecutores del clúster.
  • La autenticación de cuadernos de Azure Databricks no funciona con los perfiles de configuración de Azure Databricks.

Si quieres llamar a las API de nivel de cuenta de Azure Databricks o si deseas usar un tipo de autenticación de Databricks distinto de la autenticación predeterminada de Databricks Notebook, también se admiten los siguientes tipos de autenticación:

Tipo de autenticación Versiones del SDK de Databricks para Python
autenticación de máquina a máquina (M2M) de OAuth 0.18.0 y versiones posteriores
Autenticación de usuario a máquina (U2M) de OAuth 0.19.0 y versiones posteriores
Autenticación de entidad de servicio de Id. de Microsoft Entra Todas las versiones
Autenticación de la CLI de Azure Todas las versiones
Autenticación de token de acceso personal de Databricks Todas las versiones

Aún no se admite la autenticación de identidades administradas de Azure.

Paso 1: Instalación o actualización del SDK de Databricks para Python

  1. Los cuadernos de Python de Azure Databricks pueden usar el SDK de Databricks para Python como cualquier otra biblioteca de Python. Para instalar o actualizar el SDK de Databricks para Python en el clúster de Azure Databricks asociado, ejecute el comando magic %pip desde una celda del cuaderno de la manera siguiente:

    %pip install databricks-sdk --upgrade
    
  2. Después de ejecutar el comando magic %pip, debe reiniciar Python a fin de que la biblioteca instalada o actualizada esté disponible para el cuaderno. Para ello, ejecute el siguiente comando desde una celda de cuaderno inmediatamente después de la celda con el %pip comando magic:

    dbutils.library.restartPython()
    
  3. Para mostrar la versión instalada del SDK de Databricks para Python, ejecute el siguiente comando desde una celda del cuaderno:

    %pip show databricks-sdk | grep -oP '(?<=Version: )\S+'
    

Paso 2: Ejecución de la aplicación

En las celdas del cuaderno, cree código de Python que importe y, a continuación, llame al SDK de Databricks para Python. En el ejemplo siguiente se usa la autenticación predeterminada del cuaderno de Azure Databricks para enumerar todos los clústeres del área de trabajo de Azure Databricks:

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

for c in w.clusters.list():
  print(c.cluster_name)

Al ejecutar esta celda, aparece una lista de los nombres de todos los clústeres disponibles en el área de trabajo de Azure Databricks.

Para usar otro tipo de autenticación de Azure Databricks, vea Métodos de autenticación de Azure Databricks y haga clic en el vínculo correspondiente para obtener detalles técnicos adicionales.

Uso de utilidades de Databricks

Puede llamar a referencia de utilidades de Databricks (dbutils) desde el SDK de Databricks para Python que se ejecuta en la máquina de desarrollo local o desde un cuaderno de Azure Databricks.

  • Desde la máquina de desarrollo local, las utilidades de Databricks solo tienen acceso a los grupos de comandos dbutils.fs, dbutils.secrets, dbutils.widgetsy dbutils.jobs.
  • Desde un cuaderno de Azure Databricks conectado a un clúster de Azure Databricks, las utilidades de Databricks tiene acceso a todos los grupos de comandos de Databricks Utilities disponibles, no solo a dbutils.fs, dbutils.secrets y dbutils.widgets. Además, el grupo de comandos dbutils.notebook está limitado solo a dos niveles de comandos, por ejemplo dbutils.notebook.run o dbutils.notebook.exit.

Para llamar a las utilidades de Databricks desde el equipo de desarrollo local o un cuaderno de Azure Databricks, utilice dbutils en WorkspaceClient. En este ejemplo de código se usa la autenticación predeterminada del cuaderno de Azure Databricks para llamar a dbutils dentro de WorkspaceClient a fin de enumerar las rutas de acceso de todos los objetos de la raíz de DBFS del área de trabajo.

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()
d = w.dbutils.fs.ls('/')

for f in d:
  print(f.path)

Como alternativa, puede llamar directamente a dbutils. Pero solo se limita al uso de la autenticación predeterminada de Azure Databricks. En este ejemplo de código se llama directamente a dbutils para enumerar todos los objetos en la raíz DBFS del área de trabajo.

from databricks.sdk.runtime import *

d = dbutils.fs.ls('/')

for f in d:
  print(f.path)

Para acceder a los Volúmenes de Unity Catalog, use files en WorkspaceClient. Vea Administrar archivos en volúmenes de Unity Catalog. No puede usar dbutils por sí mismo ni dentro de WorkspaceClient para acceder a los volúmenes.

Consulte también Interacción con dbutils.

Ejemplos de código

En los ejemplos de código siguientes se muestra cómo usar el SDK de Databricks para Python para crear y eliminar clústeres, crear trabajos y enumerar los grupos a nivel de cuenta. En los ejemplos de código se usa la autenticación predeterminada de cuadernos de Azure Databricks. Para más información sobre la autenticación de cuadernos predeterminada de Azure Databricks, vea Uso del SDK de Databricks para Python desde un cuaderno de Azure Databricks. Para más información sobre la autenticación predeterminada fuera de los cuadernos, consulte Autenticación del SDK de Databricks para Python con la cuenta o el área de trabajo de Azure Databricks.

Para obtener ejemplos de código adicionales, consulte los ejemplos en el repositorio del SDK de Databricks para Python en GitHub. Consulte también:

Crear un clúster

En este ejemplo de código se crea un clúster con la versión de Databricks Runtime y el tipo de nodo de clúster especificados. Este clúster tiene un trabajo y el clúster se finalizará automáticamente después un tiempo de inactividad de 15 minutos.

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

print("Attempting to create cluster. Please wait...")

c = w.clusters.create_and_wait(
  cluster_name             = 'my-cluster',
  spark_version            = '12.2.x-scala2.12',
  node_type_id             = 'Standard_DS3_v2',
  autotermination_minutes  = 15,
  num_workers              = 1
)

print(f"The cluster is now ready at " \
      f"{w.config.host}#setting/clusters/{c.cluster_id}/configuration\n")

Eliminación permanente de un usuario

En este ejemplo de código, se elimina permanentemente el clúster con el identificador de clúster especificado del área de trabajo.

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

c_id = input('ID of cluster to delete (for example, 1234-567890-ab123cd4): ')

w.clusters.permanent_delete(cluster_id = c_id)

Creación de un trabajo

En este ejemplo de código, se crea un trabajo de Azure Databricks que ejecuta el cuaderno especificado en el clúster especificado. A medida que se ejecuta el código, obtiene la ruta de acceso del cuaderno existente, el identificador de clúster existente y la configuración del trabajo relacionado del usuario en el terminal.

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import Task, NotebookTask, Source

w = WorkspaceClient()

job_name            = input("Some short name for the job (for example, my-job): ")
description         = input("Some short description for the job (for example, My job): ")
existing_cluster_id = input("ID of the existing cluster in the workspace to run the job on (for example, 1234-567890-ab123cd4): ")
notebook_path       = input("Workspace path of the notebook to run (for example, /Users/someone@example.com/my-notebook): ")
task_key            = input("Some key to apply to the job's tasks (for example, my-key): ")

print("Attempting to create the job. Please wait...\n")

j = w.jobs.create(
  name = job_name,
  tasks = [
    Task(
      description = description,
      existing_cluster_id = existing_cluster_id,
      notebook_task = NotebookTask(
        base_parameters = dict(""),
        notebook_path = notebook_path,
        source = Source("WORKSPACE")
      ),
      task_key = task_key
    )
  ]
)

print(f"View the job at {w.config.host}/#job/{j.job_id}\n")

Creación de un trabajo que use proceso sin servidor

El siguiente ejemplo crea un trabajo que utiliza Computación sin servidor para trabajos:

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.jobs import NotebookTask, Source, Task

w = WorkspaceClient()

j = w.jobs.create(
  name = "My Serverless Job",
  tasks = [
    Task(
      notebook_task = NotebookTask(
      notebook_path = "/Users/user@databricks.com/MyNotebook",
      source = Source("WORKSPACE")
      ),
      task_key = "MyTask",
   )
  ]
)

Administración de archivos en volúmenes del Unity Catalog

En este ejemplo de código se muestran varias llamadas a files funcionalidad dentro de WorkspaceClient para acceder a un volumen de Unity Catalog.

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# Define volume, folder, and file details.
catalog            = 'main'
schema             = 'default'
volume             = 'my-volume'
volume_path        = f"/Volumes/{catalog}/{schema}/{volume}" # /Volumes/main/default/my-volume
volume_folder      = 'my-folder'
volume_folder_path = f"{volume_path}/{volume_folder}" # /Volumes/main/default/my-volume/my-folder
volume_file        = 'data.csv'
volume_file_path   = f"{volume_folder_path}/{volume_file}" # /Volumes/main/default/my-volume/my-folder/data.csv
upload_file_path   = './data.csv'

# Create an empty folder in a volume.
w.files.create_directory(volume_folder_path)

# Upload a file to a volume.
with open(upload_file_path, 'rb') as file:
  file_bytes = file.read()
  binary_data = io.BytesIO(file_bytes)
  w.files.upload(volume_file_path, binary_data, overwrite = True)

# List the contents of a volume.
for item in w.files.list_directory_contents(volume_path):
  print(item.path)

# List the contents of a folder in a volume.
for item in w.files.list_directory_contents(volume_folder_path):
  print(item.path)

# Print the contents of a file in a volume.
resp = w.files.download(volume_file_path)
print(str(resp.contents.read(), encoding='utf-8'))

# Delete a file from a volume.
w.files.delete(volume_file_path)

# Delete a folder from a volume.
w.files.delete_directory(volume_folder_path)

Enumeración de grupos a nivel de cuenta

En este ejemplo de código se enumeran los nombres para mostrar de todos los grupos disponibles en la cuenta de Azure Databricks.

from databricks.sdk import AccountClient

a = AccountClient()

for g in a.groups.list():
  print(g.display_name)

Prueba

Para probar el código, use marcos de pruebas de Python como pytest. Para probar el código en condiciones simuladas sin llamar a los puntos de conexión de la API de REST de Azure Databricks ni cambiar el estado de las cuentas o áreas de trabajo de Azure Databricks, use bibliotecas de simulación de Python como unittest.mock.

Sugerencia

Databricks Labs proporciona un complemento pytest para simplificar las pruebas de integración con Databricks y un complemento pylint para garantizar la calidad del código.

El siguiente archivo de ejemplo denominado helpers.py contiene una create_cluster función que devuelve información sobre el nuevo clúster:

# helpers.py

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.compute import ClusterDetails

def create_cluster(
  w: WorkspaceClient,
  cluster_name:            str,
  spark_version:           str,
  node_type_id:            str,
  autotermination_minutes: int,
  num_workers:             int
) -> ClusterDetails:
  response = w.clusters.create(
    cluster_name            = cluster_name,
    spark_version           = spark_version,
    node_type_id            = node_type_id,
    autotermination_minutes = autotermination_minutes,
    num_workers             = num_workers
  )
  return response

Dado el siguiente archivo denominado main.py que llama a la create_cluster función :

# main.py

from databricks.sdk import WorkspaceClient
from helpers import *

w = WorkspaceClient()

# Replace <spark-version> with the target Spark version string.
# Replace <node-type-id> with the target node type string.
response = create_cluster(
  w = w,
  cluster_name            = 'Test Cluster',
  spark_version           = '<spark-version>',
  node_type_id            = '<node-type-id>',
  autotermination_minutes = 15,
  num_workers             = 1
)

print(response.cluster_id)

El siguiente archivo denominado test_helpers.py comprueba si la función create_cluster devuelve la respuesta esperada. En lugar de crear un clúster en el área de trabajo de destino, esta prueba simula un objeto WorkspaceClient, define la configuración del objeto simulado y después pasa el objeto simulado a la función create_cluster. A continuación, la prueba comprueba si la función devuelve el nuevo identificador esperado del clúster ficticio.

# test_helpers.py

from databricks.sdk import WorkspaceClient
from helpers import *
from unittest.mock import create_autospec # Included with the Python standard library.

def test_create_cluster():
  # Create a mock WorkspaceClient.
  mock_workspace_client = create_autospec(WorkspaceClient)

  # Set the mock WorkspaceClient's clusters.create().cluster_id value.
  mock_workspace_client.clusters.create.return_value.cluster_id = '123abc'

  # Call the actual function but with the mock WorkspaceClient.
  # Replace <spark-version> with the target Spark version string.
  # Replace <node-type-id> with the target node type string.
  response = create_cluster(
    w = mock_workspace_client,
    cluster_name            = 'Test Cluster',
    spark_version           = '<spark-version>',
    node_type_id            = '<node-type-id>',
    autotermination_minutes = 15,
    num_workers             = 1
  )

  # Assert that the function returned the mocked cluster ID.
  assert response.cluster_id == '123abc'

Para ejecutar esta prueba, ejecute el comando pytest desde la raíz del proyecto de código, que debe generar resultados de prueba similares a los siguientes:

$ pytest
=================== test session starts ====================
platform darwin -- Python 3.12.2, pytest-8.1.1, pluggy-1.4.0
rootdir: <project-rootdir>
collected 1 item

test_helpers.py . [100%]
======================== 1 passed ==========================

Recursos adicionales

Para más información, consulte: